"""StepRunner 베이스 클래스 — 모든 파이프라인 단계의 실행 프레임워크. 각 단계는 이 클래스를 상속하여 _execute()만 구현하면 됨. 게이트, 체크포인트, 무효화, Opik 추적을 자동 처리. """ import hashlib import json import logging import os import re import uuid from dataclasses import dataclass from datetime import datetime, timezone from enum import Enum from pathlib import Path from typing import Any, Dict, List, Optional, TYPE_CHECKING from sqlalchemy import text from sqlalchemy.orm import Session as OrmSession if TYPE_CHECKING: from app.core.integrity_report import CompletionReport, CleanupReport from app.core.checkpoint_io import atomic_write_json from app.core.errors import AppError # W1-F10: STEP_MANIFEST dict 직참조 제거. 함수 API만 사용. # step_catalog는 STEP_CLASSES 바인딩(소비자 전용 view)이고, step_runner는 그 구축 대상이라 # 순환을 피하기 위해 함수 원형을 step_manifest에서 직접 import한다. from app.core.step_manifest import ( contains as _step_contains, get_manifest_dict, get_depends_on, get_all_downstream_recursive, ) logger = logging.getLogger(__name__) # Audit A4 (DRY): recovery loop 무한 차단 한계. dual gate(resume + force 분기)를 # 하나로 통합하기 위해 module-level 상수로 추출. 두 분기 모두 _record_recovery # 직후 _check_recovery_exhausted를 호출하면 동일 동작이 보장된다. MAX_RECOVERY_ATTEMPTS: int = 3 # Block B T1~T3 (plan v2.1.3 / spec V5 §2.3): resume 판정 모델. # # 본 commit scope: type 도입 + observation helper. run() 의 SKIP/BLOCK 분기만 # 디스패치 도입, RERUN_SELF/FORCE_EXPLICIT 부수효과는 기존 그대로 보존. # 실제 정책 변경 (running 자동 force 금지 / verify_crashed origin 분리 / # rerun_self vs force 실행 분리) 은 후속 task (B4 / B7 / B12) 에서. class ResumeAction(Enum): """ResumeDecision.action — claim 후 실행 분기 결정 (V2 patch I1, spec §2.3).""" SKIP = "skip" RERUN_SELF = "rerun_self" FORCE_EXPLICIT = "force_explicit" STALE_RUNNING_RECOVERY = "stale_running_recovery" BLOCK = "block" NOT_APPLICABLE = "not_applicable" @dataclass(frozen=True) class ResumeDecision: """판정 결과 + 사유. V3 patch B1: STALE_RUNNING_RECOVERY 시 expected_started_at + expected_run_id 동반. claim SQL 의 atomic steal 조건에서 SQL cast 없이 (text 비교만) 검증. `origin` 은 caller (run()) 가 부수효과 디스패치에 사용: - None: first-run 또는 unknown-status (cleanup 없이 execute) - "artifact_missing": completed-mismatch (record_recovery + check_exhausted + cleanup) - "prior_state": non-completed prior status (force-like log + cleanup) 후속 task (B11) 에서 CompletionReport.origin 과 통합. """ action: ResumeAction reason: str origin: Optional[str] = None expected_started_at: Optional[str] = None expected_run_id: Optional[str] = None def compute_config_hash(project_config: Optional[Dict]) -> str: """project_config의 결정적 해시. P0-3 (Codex H2) — resume 안전성용. 같은 config면 같은 hash → 체크포인트가 만들어진 시점의 config와 현재 config가 다르면 mismatch 감지 가능. default fallback은 결정적이어야 한다. `default=str`을 쓰면 같은 클래스의 다른 인스턴스(예: ``)가 매번 다른 hash를 만들어 silent false-mismatch 회귀를 유발한다. 비-JSON 타입은 클래스명만 기록. Returns 16자 md5 prefix. """ canonical = json.dumps( project_config or {}, sort_keys=True, ensure_ascii=False, default=lambda o: f"", ) return hashlib.md5(canonical.encode("utf-8")).hexdigest()[:16] class StepRunner: """파이프라인 단계 실행기. 서브클래스는 _execute()만 구현. run()이 게이트 → 체크포인트 → 실행 → 저장 → 무효화를 자동 처리. """ def __init__( self, step_id: str, project_id: str, episode_id: str, db: OrmSession, project_config: Optional[Dict] = None, opik_context: Optional[Dict] = None, ): if not _step_contains(step_id): raise ValueError(f"Unknown step: {step_id}") self.step_id = step_id self.project_id = project_id self.episode_id = episode_id self.db = db self.project_config = project_config or {} self.manifest = get_manifest_dict(step_id) self.run_id = str(uuid.uuid4()) self.opik_context = opik_context or {} # 체크포인트 경로 from app.core.config import settings self._cp_dir = ( Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / step_id ) # ── 게이트 ── def check_gate(self) -> None: """의존 단계 완료 확인. blocked이면 AppError. partial cascade contract (problems.md #11 + G4.6 Phase 3 fix iter 2): - dep status == 'partial' 이고 dep manifest 의 ``allow_partial_downstream`` 가 ``False`` 면 blocked. 명시 contract 없는 dep (default True) 는 backward-compat 으로 통과. - 즉 emitter (dep step) 의 출력 contract 에 따라 cascade 제어. - **Operator override (manifest-driven)**: dep manifest 의 ``partial_override_config_key`` 가 명시되어 있고 project_config 의 동일 키 값이 truthy 면, partial 차단을 통과 (운영자가 명시적으로 partial 진행을 결정한 경우). WARNING log 동시. """ blocked_by = [] for dep in get_depends_on(self.step_id): dep_run = self._get_step_run(dep) if not dep_run: blocked_by.append(dep) continue dep_status = dep_run["status"] if dep_status in ("completed", "not_applicable"): continue if dep_status == "partial": dep_meta = get_manifest_dict(dep) or {} if dep_meta.get("allow_partial_downstream", True) is False: override_key = dep_meta.get("partial_override_config_key") # strict — `is True` 만 허용. "true" / "1" / 1 같은 truthy # 표현은 명시적 bool 결정이 아니므로 차단. if override_key and self.project_config.get(override_key) is True: logger.warning( "check_gate: dep %s partial — explicit operator " "override via project_config['%s']=True. proceeding.", dep, override_key, ) continue blocked_by.append(f"{dep}(partial,strict)") # else: backward-compat 통과 continue blocked_by.append(f"{dep}({dep_status})") if blocked_by: dep_labels = [(get_manifest_dict(b.split("(")[0]) or {}).get("label", b) for b in blocked_by] raise AppError( code="gate.blocked", message=f"선행 단계 미완료: {', '.join(dep_labels)}", status_code=400, ) def check_applicability(self) -> bool: """이 단계가 적용 가능한지. False면 not_applicable. 기본 구현은 `app.core.applicability.resolve_applicability`를 호출. 특수 로직이 필요한 서브클래스(SceneSplitStep, SceneVerifyStep 등)는 override. """ from app.core.applicability import resolve_applicability return resolve_applicability(self) # ── 체크포인트 ── # archive 파일명 패턴: manifest_YYYYMMDD_HHMMSS[_].json _ARCHIVE_PATTERN = re.compile( r"^manifest_(\d{8}_\d{6})(?:_[a-zA-Z0-9_-]+)?\.json$" ) # F4-1: clear_checkpoint(force trigger) 시 작성하는 marker. # archive fallback이 force 의도(이전 cp 폐기)를 무효화하는 사고를 막는다. # save_checkpoint(completed)가 정상 진행 시 제거. _FORCE_CLEARED_MARKER = ".force_cleared" def _is_force_cleared(self) -> bool: """force_cleared marker 존재 여부 — load_checkpoint이 archive/manifest 모두 무효 처리하는 분기.""" return (self._cp_dir / self._FORCE_CLEARED_MARKER).exists() def load_checkpoint(self) -> Optional[Dict]: """체크포인트 manifest.json 로드. Primary: manifest.json 직접 로드. Fallback (Fix 4 / M5): manifest.json이 빈 파일/없음/손상이면 가장 최근 archive(`manifest_TIMESTAMP[_runid].json`)로 자동 복원. archive가 살아있는 한 force 직후 backend kill 같은 사고에서도 dispatcher가 cp를 회복한다. F4-1 (Claude IMPORTANT): clear_checkpoint 직후 backend kill로 save 누락된 시나리오에선 archive(이전 completed)가 force 의도를 silent로 무효화할 수 있다. clear_checkpoint이 작성한 `.force_cleared` marker를 확인하여 archive 복구 + manifest.json 복원을 모두 차단 — resume이 stale로 인식하여 재실행. save_checkpoint(completed)가 정상 진행되면 marker 자동 제거. P2-1 (Codex): archive가 status='running' incremental snapshot이면 복구 후보에서 제외 — `_find_latest_archive`가 completed/partial archive만 반환. archive → manifest 자동 복구는 atomic_write_json으로 기록하므로 재시도 호출에서 fallback 비용이 1회만 발생. """ manifest_path = self._cp_dir / "manifest.json" force_cleared = self._is_force_cleared() # Primary: manifest.json 직접 로드 (빈 파일/손상은 fallback으로 진입). if manifest_path.exists(): try: text_data = manifest_path.read_text(encoding="utf-8").strip() if text_data: data = json.loads(text_data) if force_cleared: # F4-1: marker가 있으면 force 의도 — archive로 복원된 manifest일 가능성. # 이전 completed cp를 stale로 처리하여 force 진행 보장. logger.warning( "Step %s: manifest restored but .force_cleared marker exists — " "treating as stale (force intent honored)", self.step_id, ) return None return data logger.warning( "Checkpoint manifest.json empty for %s. Trying archive fallback.", self.step_id, ) except Exception as exc: logger.warning( "Checkpoint load failed for %s: %s. Trying archive fallback.", self.step_id, exc, ) # F4-1: force_cleared marker 있으면 archive 복원 자체를 차단. if force_cleared: logger.warning( "Step %s: .force_cleared marker present — skipping archive fallback " "(force intent honored, will rerun)", self.step_id, ) return None # Fallback: 최신 archive 자동 복원. archive = self._find_latest_archive() if archive is None: return None try: data = json.loads(archive.read_text(encoding="utf-8")) except Exception as exc: logger.error( "Archive fallback parse failed for %s (archive=%s): %s", self.step_id, archive.name, exc, ) return None # archive → manifest.json 복구. 다음 load 호출은 primary 경로로 처리됨. try: atomic_write_json(manifest_path, data) logger.warning( "Step %s: manifest.json missing/empty/corrupt. Auto-restored from archive %s", self.step_id, archive.name, ) except Exception as exc: # 복구 기록 실패해도 데이터 자체는 archive에서 읽었으므로 반환. logger.error( "Archive fallback write failed for %s (archive=%s): %s", self.step_id, archive.name, exc, ) return data def _find_latest_archive(self) -> Optional[Path]: """최신 manifest_TIMESTAMP.json archive 파일 경로 반환 (없으면 None). 파일명 패턴: manifest_YYYYMMDD_HHMMSS[_].json TIMESTAMP 기준 내림차순으로 정렬 후, status='completed' 또는 'partial' 인 archive만 후보로 채택 (P2-1 Codex fix). P2-1: incremental save_checkpoint(status='running') archive를 복원하면 step_run.status='completed'로 남고 cp는 running snapshot이 되어 sync 누락 + downstream silent skip 사고 발생. final 상태 archive만 복구. 손상된 archive는 skip 후 다음 후보 시도. """ if not self._cp_dir.exists(): return None archives: list[tuple[str, Path]] = [] for p in self._cp_dir.iterdir(): if not p.is_file(): continue m = self._ARCHIVE_PATTERN.match(p.name) if m: archives.append((m.group(1), p)) if not archives: return None archives.sort(key=lambda x: x[0], reverse=True) # 최신부터 순회하며 final 상태(completed/partial)인 archive만 선택. # status 필드 부재(legacy archive)는 final로 간주 — 보수적 호환. for ts, path in archives: try: payload = json.loads(path.read_text(encoding="utf-8")) except Exception as exc: logger.warning( "Archive parse failed (skip): %s — %s", path.name, exc ) continue status = payload.get("status") if isinstance(payload, dict) else None if status is None or status in ("completed", "partial", "not_applicable"): return path logger.debug( "Archive skipped (status=%s, not final): %s", status, path.name ) return None @staticmethod def _archive_manifest(manifest_path: "Path") -> None: """기존 manifest.json을 날짜시간 버전으로 복사 보관. 원본은 그대로 유지.""" if not manifest_path.exists(): return try: import shutil from datetime import datetime ts = datetime.now().strftime("%Y%m%d_%H%M%S") archive_path = manifest_path.parent / f"manifest_{ts}.json" # 유일성 보장: run_id 8자리 추가 if archive_path.exists(): import uuid archive_path = manifest_path.parent / f"manifest_{ts}_{uuid.uuid4().hex[:8]}.json" shutil.copy2(str(manifest_path), str(archive_path)) logger.debug("Archived checkpoint: %s", archive_path.name) except Exception as exc: logger.warning("Checkpoint archive failed: %s", exc) def save_checkpoint(self, data: Dict) -> None: """체크포인트 저장 (원자적). 기존 manifest.json은 복사로 보관 후 덮어쓰기. F4-1: status='completed' 또는 'partial' 저장 시 .force_cleared marker 제거 — force 명령이 정상 진행 완료된 신호. 'running' incremental save에서는 marker를 유지하여 backend kill 시 force 의도 보존. """ manifest_path = self._cp_dir / "manifest.json" data["step_id"] = self.step_id data["run_id"] = self.run_id data["resolved_model"] = self._resolve_model() data["updated_at"] = self._now() # P0-3 (Codex H2): resume 안전성용 메타. schema_version은 manifest에서 # step별 정의 (default 1, 체크포인트 형식 변경 시 +1). # config_hash mismatch는 step_execution_service가 resume 거부 트리거. # Phase 5.1: step이 _execute()에서 schema_version/config_hash를 명시적으로 # 반환했다면(PROMPT_VERSION 등 step-local 시그널 포함) 그 값을 보존한다. # 누락 시에만 manifest default + project_config-only hash로 fallback. if "schema_version" not in data: data["schema_version"] = self.manifest.get("schema_version", 1) if "config_hash" not in data: data["config_hash"] = compute_config_hash(self.project_config) # D2 fix: project_config_snapshot 기록 — _diff_project_config가 None reference로 # 폴백되어 매번 stale auto-rerun trigger되는 패턴 종결. # 민감 키(api_key/password/token/secret/credential)는 마스킹. # step이 _execute()에서 명시 제공한 snapshot이 있으면 보존(Phase 5.1 패턴 일관). if "project_config_snapshot" not in data: data["project_config_snapshot"] = self._mask_sensitive_keys(self.project_config) try: # 기존 manifest는 아카이브 → atomic_write_json (tmp → rename) self._archive_manifest(manifest_path) atomic_write_json(manifest_path, data) except Exception as exc: logger.error("Checkpoint save failed for %s: %s", self.step_id, exc) raise RuntimeError(f"Checkpoint save failed for {self.step_id}: {exc}") from exc # F4-1: 정상 진행(completed/partial/not_applicable) 시 marker 제거 — best-effort. # 'running' incremental save에서는 유지 (backend kill 시 force 의도 살아있음). final_status = data.get("status") if final_status in ("completed", "partial", "not_applicable"): marker = self._cp_dir / self._FORCE_CLEARED_MARKER try: if marker.exists(): marker.unlink(missing_ok=True) except Exception as exc: # marker 제거 실패는 fix 자체를 깨뜨리지 않음 — 다음 force 시 덮어쓰기. logger.warning( "Failed to clear .force_cleared marker for %s: %s", self.step_id, exc, ) def clear_checkpoint(self) -> None: """체크포인트 삭제 (기존 파일은 날짜시간 버전으로 보관 후 삭제). F4-1: clear는 force trigger의 일부 — 이후 backend kill로 save_checkpoint가 호출되지 않더라도 force 의도를 보존하기 위해 .force_cleared marker 작성. load_checkpoint이 이를 감지하면 archive/manifest 복구를 모두 차단하여 step을 stale 상태로 유지 → 다음 실행 시 force 동작 보장. marker 작성은 best-effort — 실패해도 clear 자체는 진행. 다음 정상 save에서 marker는 자동 제거된다. """ manifest_path = self._cp_dir / "manifest.json" if manifest_path.exists(): self._archive_manifest(manifest_path) manifest_path.unlink(missing_ok=True) # F4-1: marker 작성 (cp_dir 보장 + best-effort touch). try: self._cp_dir.mkdir(parents=True, exist_ok=True) (self._cp_dir / self._FORCE_CLEARED_MARKER).touch(exist_ok=True) except Exception as exc: # marker 작성 실패는 fix 자체를 깨뜨리지 않음 — 기존 동작(archive 복원 가능) 그대로. logger.warning( "Failed to write .force_cleared marker for %s: %s", self.step_id, exc, ) # ── 무효화 ── def invalidate_downstream(self, target_step_id: Optional[str] = None, *, delete_checkpoints: bool = True) -> None: """downstream step을 stale 처리 + 체크포인트 삭제 (또는 stale 플래그만). Args: target_step_id: None이면 self.step_id 기준. editorial step이 타 step 체크포인트를 덮어쓴 경우 해당 step_id를 전달하여 그의 downstream을 무효화. delete_checkpoints: True면 체크포인트 파일까지 아카이브+삭제. False면 DB step_run만 stale (체크포인트는 덮어쓴 값 그대로 유지, editorial 재실행 용). """ ref = target_step_id or self.step_id downstream = get_all_downstream_recursive(ref) # 2-pass 의존성: ref가 역참조로 소비하는 하류는 파일 보존(DB는 stale). # step_manifest의 depends_on은 단방향 DAG이지만, 일부 상류가 하류 체크포인트를 # 역참조로 소비하는 구조가 있다. 예: scene_detail.consumes_downstream = # ["shot_dependency_t2i"] — scene_context_loader가 refined ref_usage(zoom_in_detail # 등)를 읽어 user_prompt 주입. ref force 시 해당 파일을 지우면 dead code. # 보존 시 DB는 stale로 남으므로 **사용자는 force 완료 후 선언된 하류도 수동 재실행**해야 # 낡은 ref_usage drift를 방지할 수 있다 (로그로 알림). # 다른 경로(예: 이 하류의 직접 depends_on 상위)가 force되면 일반 삭제 — 의미 명시적. # catalog 헬퍼 경유 — step_catalog 설계 원칙(소비자는 STEP_CATALOG만 본다) 준수. # 지연 import: step_catalog._build()가 app.core.steps → step_runner를 로드하는 순환 회피. from app.core.step_catalog import get_consumes_downstream consumes: set = set(get_consumes_downstream(ref)) now = self._now() from app.core.config import settings for sid in downstream: self.db.execute(text( "UPDATE step_run SET status = 'stale', updated_at = :now " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid " "AND status NOT IN ('pending', 'stale')" ), {"pid": self.project_id, "eid": self.episode_id, "sid": sid, "now": now}) if delete_checkpoints: if sid in consumes: logger.info( "Preserved stale checkpoint for %s (consumed by %s) — " "파일 보존, DB stale. drift 방지를 위해 %s force 완료 후 %s도 재실행 권장", sid, ref, ref, sid, ) continue # 체크포인트 보관 후 삭제 (보관 실패해도 삭제는 진행) cp_path = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / sid / "manifest.json" ) if cp_path.exists(): self._archive_manifest(cp_path) cp_path.unlink(missing_ok=True) # ★지웠다는 표식을 남긴다 (2026-08-07 실측·Codex 3차). # # 표식이 없으면 `load_checkpoint` 의 archive 자동 복원이 # 방금 **의도적으로 지운 산출을 되살려** 현재 입력과 # 섞는다. 자기 force 로 지우는 `clear_checkpoint` 는 # 이미 같은 표식을 쓰는데, 상류 force 가 하류를 지우는 # 이 경로만 빠져 있었다. # # 실측: `entity_t2i` 가 복원된 옛 `completed` 를 현재 # 대상과 대조 없이 받아 partial(154/97, failed=-57)이 # 됐고, 그 때문에 entity sync 가 stale cleanup 을 미룬 # 채 INSERT 해 UniqueViolation 으로 `scene_director` # 앞 필수 사전 동기화가 실패, 전체가 20분간 멈췄다. # 경합이 아니라 **결정론적 stale 부활**이다. try: cp_path.parent.mkdir(parents=True, exist_ok=True) (cp_path.parent / self._FORCE_CLEARED_MARKER).touch(exist_ok=True) except Exception as exc: # noqa: BLE001 # 표식 실패가 무효화 자체를 깨뜨리지는 않는다 — # 다만 그 스텝은 복원 위험이 남으므로 경고로 남긴다. logger.warning( "Failed to mark %s as invalidated (archive 복원 " "위험 잔존): %s", sid, exc, ) if downstream: self.db.commit() logger.info( "Invalidated %d downstream steps from %s (delete_cp=%s): %s", len(downstream), ref, delete_checkpoints, downstream, ) # ── DB step_run 관리 ── def _get_step_run(self, step_id: str) -> Optional[Dict[str, Any]]: """step_run row 의 모든 결정 필드를 dict 로 반환. Block B T0 (plan v2.1.3 / spec V5 S7): atomic claim / stale steal / verify path 가 run_id + started_at + recovery_count 의존 — tuple index 오해석 위험을 0으로 만든다. tuple fallback 절대 도입 X. """ row = self.db.execute(text(""" SELECT status, run_id, started_at, completed_count, applicable_count, COALESCE(recovery_count, 0) AS recovery_count, updated_at FROM step_run WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid """), {"pid": self.project_id, "eid": self.episode_id, "sid": step_id}).fetchone() if row is None: return None return { "status": row.status, "run_id": row.run_id, "started_at": row.started_at, "completed_count": row.completed_count, "applicable_count": row.applicable_count, "recovery_count": row.recovery_count, "updated_at": row.updated_at, } # ── ResumeDecision 판정 helper (Block B T1~T3 + B11, spec V5 §2.3) ── # # helper 의 직접 부수효과는 0 — `_record_recovery` / `_update_step_run` / # `logger.warning` / 자체 `raise` 없음. recovery / log / AppError 디스패치는 # caller (run()) 책임. # # 다만 의존 함수의 기존 side effect 는 호출: # - `load_checkpoint()`: archive fallback 시 manifest.json rewrite # - `_safe_verify_completion()` (B7): unexpected exception 시 AppError( # step.verify_crashed) raise — helper 내부에서 catch 안 함, 자연 propagate. # AppError(다른 code) 는 CompletionReport(origin='contract_drift') 로 변환. # # Block C (claim 전 non-mutating decision) 전제 와의 잠재 충돌 (load_checkpoint # 의 archive rewrite) 은 후속 task 에서 별도 정리. # # B11 scope (이번 commit): verify-derived contract_drift / invariant_drift 시 # BLOCK 디스패치. cp_mismatch (schema/config_hash) 의 BLOCK 정책은 B5 # (`_LEGACY_SCHEMA_BUMP_ALLOWLIST` + `_evaluate_contract_drift`) follow-up. def _evaluate_resume_decision(self, mode: str = "resume") -> ResumeDecision: """resume 판정자 — `(mode, step_run row, cp, verify_completion, project_config)` 를 관찰해 `ResumeDecision` 반환. 본 helper 는 자체 raise 안 함. `_safe_verify_completion` (B7) 이 AppError(step.verify_crashed) raise 시 helper 내부에서 catch 안 함 — caller (run() / step_execution_service) 가 fail-fast 처리. 분기 매핑 (current run() 동작 1:1 보존): - mode='force' → FORCE_EXPLICIT - first-run (no row) → RERUN_SELF (origin=None — cleanup 없음) - status='completed': - cp clean + verify pass → SKIP - cp None / cp mismatch / verify fail: - strict_resume=True + cp not None → BLOCK - 그 외 → RERUN_SELF (origin='artifact_missing') - status in {running, failed, partial, stale, pending}: - RERUN_SELF (origin='prior_state') - NOTE: running 자동 force 정책 변경은 후속 task (B4 STALE_RUNNING_RECOVERY). - 그 외 unknown status → RERUN_SELF (origin=None — current run() 의 fall-through 동작 보존) BLOCK 정책은 후속 task (B11) 에서. """ if mode == "force": return ResumeDecision( action=ResumeAction.FORCE_EXPLICIT, reason="user-requested force", ) existing = self._get_step_run(self.step_id) if not existing: return ResumeDecision( action=ResumeAction.RERUN_SELF, reason="first-run (no step_run row)", origin=None, ) status = existing["status"] if status == "completed": cp = self.load_checkpoint() # Case 1: cp 부재 — D1 사고 패턴 (PID 0bb48ebf), artifact_missing. # strict_resume 미적용 (legacy: "cp 있을 때만 의미"). if cp is None: return ResumeDecision( action=ResumeAction.RERUN_SELF, reason="checkpoint missing but step_run.status=completed", origin="artifact_missing", ) # Case 2: cp 있음, structural mismatch (schema/config_hash) 검사. cp_mismatch = self._check_cp_mismatch(cp) if cp_mismatch: # legacy strict_resume override — operator 가 명시한 경우 contract # policy 보다 우선. raw mismatch 를 reason 으로 (legacy message 보존). if self.project_config.get("strict_resume", False): return ResumeDecision( action=ResumeAction.BLOCK, reason=cp_mismatch, origin="strict_resume", ) # B5 (plan v2.1.3 / spec V5 §4.5): contract_drift 정책. # _LEGACY_SCHEMA_BUMP_ALLOWLIST (entity_t2i 만) 한정 RERUN_SELF. return self._evaluate_contract_drift(cp_mismatch) # Case 3: cp 있음 + structurally OK → verify 실행. verify_report = self._safe_verify_completion() if verify_report.is_complete: return ResumeDecision( action=ResumeAction.SKIP, reason="cp clean, verify passed", ) # legacy strict_resume — verify-fail 케이스도 cover (cp 있음). if self.project_config.get("strict_resume", False): return ResumeDecision( action=ResumeAction.BLOCK, reason=f"verify failed: {verify_report.missing}", origin="strict_resume", ) # B11: verify-derived origin 디스패치. if verify_report.origin == "contract_drift": return ResumeDecision( action=ResumeAction.BLOCK, reason=( f"contract drift detected by verify: " f"{verify_report.missing[:3]}" ), origin="contract_drift", ) if verify_report.origin == "invariant_drift": # 1차 정책 (D4 까지): mutator 식별 불가 → safer default BLOCK. return ResumeDecision( action=ResumeAction.BLOCK, reason=( f"invariant drift detected by verify " f"(1차 정책): {verify_report.missing[:3]}" ), origin="invariant_drift", ) # default verify-derived RERUN_SELF (artifact_missing / clean). return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=f"verify failed: {verify_report.missing}", origin="artifact_missing", ) # B4 (plan v2.1.3 / spec V5 §4.6): running 자동 force 차단. # 이전 동작은 status='running' 도 prior_state RERUN_SELF (force-like) 로 처리 → # 다른 worker 와 동시 실행 / DB row 덮어쓰기 / lock 사고 trigger. 본 helper 가 # timeout 기반으로 BLOCK 또는 STALE_RUNNING_RECOVERY 로 분기. if status == "running": return self._evaluate_running_state(existing) if status in ("failed", "partial", "stale", "pending"): return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=f"status={status} → force-like recovery", origin="prior_state", ) # unknown status — current run() 은 fall through → execute (cleanup 없음). # 동일 동작 보존을 위해 origin=None (first-run 마커 reuse). return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=f"unknown status: {status!r}", origin=None, ) def _evaluate_running_state(self, existing: Dict[str, Any]) -> ResumeDecision: """status='running' row 평가 (Block B B4, plan v2.1.3 / spec V5 §4.6). - started_at NULL → BLOCK (정상 row 에서 발생 안 함, manual investigation) - started_at parse 실패 → BLOCK (corrupt timestamp) - elapsed < timeout → BLOCK ("healthy" — 다른 worker 가 정상 진행 중일 가능성) - elapsed >= timeout → STALE_RUNNING_RECOVERY + expected_started_at + expected_run_id (Block C atomic steal 입력 — text 비교 만으로 race-safe). 본 helper 는 자체 raise 안 함 — caller (run()) 가 BLOCK / STALE 디스패치 책임. """ from app.core.config import settings started_at_raw = existing.get("started_at") run_id_raw = existing.get("run_id") if started_at_raw is None: return ResumeDecision( action=ResumeAction.BLOCK, reason="running with started_at=None (manual investigation required)", origin="running_invalid", ) try: started_at = datetime.fromisoformat( str(started_at_raw).replace("Z", "+00:00") ) except (ValueError, TypeError, AttributeError): return ResumeDecision( action=ResumeAction.BLOCK, reason=f"running with started_at parse failed: {started_at_raw!r}", origin="running_invalid", ) # naive datetime (timezone 미포함) 방어 — UTC 가정. if started_at.tzinfo is None: started_at = started_at.replace(tzinfo=timezone.utc) timeout_seconds = settings.step_running_timeout_seconds elapsed = (datetime.now(timezone.utc) - started_at).total_seconds() if elapsed < timeout_seconds: return ResumeDecision( action=ResumeAction.BLOCK, reason=( f"running healthy (elapsed={elapsed:.0f}s < timeout={timeout_seconds}s)" ), origin="running_healthy", ) return ResumeDecision( action=ResumeAction.STALE_RUNNING_RECOVERY, reason=( f"running stale (elapsed={elapsed:.0f}s >= timeout={timeout_seconds}s)" ), expected_started_at=started_at_raw, expected_run_id=run_id_raw, ) def _evaluate_contract_drift(self, mismatch_reason: str) -> ResumeDecision: """contract_drift 분류 (Block B B5, plan v2.1.3 / spec V5 §4.5). cp_mismatch (schema_version / config_hash) 의 BLOCK 정책. 한정 allowlist (`_LEGACY_SCHEMA_BUMP_ALLOWLIST = {entity_t2i}`) 만 schema_version mismatch 시 RERUN_SELF 허용. config_hash mismatch 는 모든 step BLOCK (사용자 의도 반영 — project_config 변경은 자동 재실행 시 의도와 다른 결과 위험). 본 helper 는 자체 raise 안 함 — caller (run()) 가 BLOCK 디스패치 책임. """ from app.core.step_manifest import _LEGACY_SCHEMA_BUMP_ALLOWLIST # schema_version mismatch + allowlist → RERUN_SELF (legacy bump 호환). if "schema_version mismatch" in mismatch_reason: if self.step_id in _LEGACY_SCHEMA_BUMP_ALLOWLIST: return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=( f"legacy schema bump allowlist: {self.step_id} " f"({mismatch_reason})" ), origin="contract_drift", ) # 그 외 모든 cp_mismatch → BLOCK (config_hash / unknown / non-allowlist # schema_version). 사용자 명시 force / 진단 강제. return ResumeDecision( action=ResumeAction.BLOCK, reason=mismatch_reason, origin="contract_drift", ) def _update_step_run( self, status: str, *, require_owner: bool = False, completed_count: int = 0, applicable_count: int = 0, failed_count: int = 0, error_message: str = "", result_summary: str = "", ) -> bool: """step_run upsert. Block C B9 (plan v2.1.3 / spec V5 §4.7, AC-C7): require_owner=True 시 ON CONFLICT DO UPDATE 의 WHERE 절에 `step_run.run_id = :rid` 추가 — 다른 worker 가 steal/overwrite 한 row 를 조용히 덮어쓰지 못하도록 차단. row 갱신/INSERT 성공 시 True, owner mismatch 시 False (RETURNING id 의 fetchone is not None 판정). """ now = self._now() started_at = now if status == "running" else None completed_at = now if status in ("completed", "not_applicable") else None owner_clause = "WHERE step_run.run_id = :rid" if require_owner else "" sql = text(f""" INSERT INTO step_run (id, project_id, episode_id, step_id, status, run_id, resolved_model, applicable_count, completed_count, failed_count, error_message, result_summary, started_at, completed_at, created_at, updated_at) VALUES (:id, :pid, :eid, :sid, :status, :rid, :model, :ac, :cc, :fc, :err, :rs, :sa, :ca, :now, :now) ON CONFLICT (project_id, episode_id, step_id) DO UPDATE SET status = :status, run_id = :rid, resolved_model = :model, applicable_count = :ac, completed_count = :cc, failed_count = :fc, error_message = :err, result_summary = :rs, updated_at = :now, started_at = COALESCE(:sa, step_run.started_at), completed_at = COALESCE(:ca, step_run.completed_at) {owner_clause} RETURNING id """) result = self.db.execute(sql, { "id": str(uuid.uuid4()), "pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "status": status, "rid": self.run_id, "model": self._resolve_model(), "ac": applicable_count, "cc": completed_count, "fc": failed_count, "err": error_message[:2000] if error_message else None, "rs": result_summary[:2000] if result_summary else None, "sa": started_at, "ca": completed_at, "now": now, }) # fetchone 은 commit 전에 — sqlite 가 commit 후 cursor 무효화 (PG 도 안전). row = result.fetchone() self.db.commit() return row is not None def _update_step_run_strict( self, status: str, *, completed_count: int = 0, applicable_count: int = 0, failed_count: int = 0, error_message: str = "", result_summary: str = "", ) -> None: """owner-aware update + False 반환 시 step.owner_lost AppError raise. V2 patch 추가 d (Codex IMPORTANT #4, plan v2.1.3 §2247): claim 후 transition 의 모든 path (success / partial / failed / exception handler / cleanup failure) 에서 owner check 실패 (False) 는 race lost 신호 — 자동 silent skip 금지. 모든 transition path 에서 surface 의무. """ updated = self._update_step_run( status, require_owner=True, completed_count=completed_count, applicable_count=applicable_count, failed_count=failed_count, error_message=error_message, result_summary=result_summary, ) if not updated: raise AppError( code="step.owner_lost", message=( f"{self.step_id} owner check failed (run_id={self.run_id}) — " f"다른 worker 가 step_run row 를 갱신했거나 stale steal 당함. " f"transition='{status}' 미적용." ), status_code=409, ) def _mark_not_applicable(self, reason: str = "") -> None: """NOT_APPLICABLE 처리 — claim 없이 DB step_run + cp 기록 (Block C, V2 patch I4). check_applicability=False 이거나 ResumeAction.NOT_APPLICABLE decision 시 호출. claim 안 했으므로 require_owner=False 로 단순 upsert. """ self._update_step_run("not_applicable", require_owner=False) self.save_checkpoint({"status": "not_applicable", "reason": reason}) def _try_claim_running( self, *, allow_stale_steal: bool = False, expected_started_at: Optional[str] = None, expected_run_id: Optional[str] = None, ) -> bool: """atomic INSERT ON CONFLICT — race-free running claim (Block C C1). AC-C1, C6, C8, C9, C10, C11 (plan v2.1.3 / spec V5 §5.2): - first-run (row 없음) → INSERT 경로, RETURNING 한 row 로 success. - 기존 row + status != 'running' → DO UPDATE 경로 (failed/partial/stale/...). - STALE_RUNNING_RECOVERY (allow_stale_steal=True) → expected-match steal: step_run.started_at = :expected_started_at AND run_id = :expected_run_id. SQL cast 0 (V3 patch B1): timeout 판정은 _evaluate_running_state() 가 Python 에서 수행. invalid text started_at row 가 있어도 query 정상 작동. fetchone() is not None 판정 (V3 patch I2): rowcount driver 차이 회피. Returns: True (claim 성공) / False (다른 worker 이미 잡음 또는 expected mismatch). """ new_step_run_id = str(uuid.uuid4()) sql = text(""" INSERT INTO step_run ( id, project_id, episode_id, step_id, run_id, status, started_at, created_at, updated_at ) VALUES ( :new_step_run_id, :pid, :eid, :sid, :run_id, 'running', :now, :now, :now ) ON CONFLICT (project_id, episode_id, step_id) DO UPDATE SET run_id = EXCLUDED.run_id, status = 'running', started_at = EXCLUDED.started_at, updated_at = EXCLUDED.updated_at, recovery_count = step_run.recovery_count + CASE WHEN step_run.status = 'running' THEN 1 ELSE 0 END WHERE step_run.status != 'running' OR ( :allow_stale_steal IS TRUE AND step_run.status = 'running' AND step_run.started_at = :expected_started_at AND step_run.run_id = :expected_run_id ) RETURNING run_id """) result = self.db.execute(sql, { "new_step_run_id": new_step_run_id, "pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "run_id": self.run_id, "now": self._now(), "allow_stale_steal": allow_stale_steal, "expected_started_at": expected_started_at, "expected_run_id": expected_run_id, }) # fetchone 은 commit 전에 — sqlite 가 commit 후 cursor 무효화 (PG 도 안전). row = result.fetchone() self.db.commit() return row is not None def update_progress(self, completed: int, total: int, failed: int = 0) -> None: """fan-out 진행률 업데이트 (running 상태에서). claim 후 owner-aware (AC-C7). Block C review I4: owner_lost 시 silent skip 대신 strict 로 즉시 raise → fan-out 중단 (partial commit orphan 차단). caller 는 outer except 에서 surface. """ self._update_step_run_strict( "running" if completed < total else "completed", completed_count=completed, applicable_count=total, failed_count=failed, ) # ── 실행 ── def validate_mode(self, mode: str) -> None: """mode 별 실행 가능성 preflight — run() 최선두(모든 mutating 단계 이전)에서 호출. subclass 가 특정 mode 를 거부해야 할 때 override 해 raise (예: 표적 씬 슬라이스+force 병용 금지 — 비표적 자산 무효화 차단, Codex 슬라이스 리뷰 BLOCKING-1). 기본=no-op.""" def run(self, mode: str = "resume") -> Dict[str, Any]: """단계 실행. mode: resume | force. Returns: {"status": "completed|skipped|not_applicable", "result": ...} Block C C2 (plan v2.1.3 / spec V5 §S2 + §S6): 4-phase 흐름 — (1) gate / applicability (non-mutating) (2) ResumeDecision 평가 (non-mutating, mutating side-effects only after claim) (3) atomic claim (RERUN_SELF / FORCE_EXPLICIT / STALE_RUNNING_RECOVERY 만) (4) origin 별 부수효과 (record_recovery / [RECOVERY] log) + _execute_* dispatch — _execute_rerun_self / _execute_force 가 _execute_and_finalize 로 finalize 까지 위임. """ # 0. mode preflight (2026-07-23 Codex 슬라이스 리뷰 BLOCKING-1) — # claim/cleanup_artifacts/invalidate_downstream/clear_checkpoint 등 # 어떤 mutating 단계보다도 먼저, subclass 가 특정 mode 실행 자체를 # fail-closed 로 거부하는 지점 (기본 no-op). self.validate_mode(mode) # 1. 게이트 확인 self.check_gate() # 2. 적용 가능 여부 (claim 전 — DB step_run row 만 마킹, race 없음) if not self.check_applicability(): self._mark_not_applicable(reason="check_applicability=False") return {"status": "not_applicable"} # 3. resume 판정 (non-mutating). Block B T1~T3 + B4 + B5 + B7+B8 + B11 + B12, # plan v2.1.3 / spec V5 §2.3. decision = self._evaluate_resume_decision(mode) if decision.action == ResumeAction.SKIP: return {"status": "skipped", "reason": "already completed"} if decision.action == ResumeAction.BLOCK: # B11: origin 별 message 포맷 (production canary 검증 운영자 friendly). if decision.origin == "strict_resume": _msg = ( f"{self.step_id} 체크포인트가 stale 입니다 " f"({decision.reason}). strict_resume=True 이므로 " "force 모드로 재실행하세요." ) elif decision.origin == "contract_drift": _msg = ( f"{self.step_id} contract drift detected — {decision.reason}. " "force 재실행 또는 수동 진단 필요." ) elif decision.origin == "invariant_drift": _msg = ( f"{self.step_id} invariant drift detected — {decision.reason}. " "force 재실행 또는 수동 진단 필요." ) elif decision.origin in ("running_healthy", "running_invalid"): _msg = ( f"{self.step_id} step is running ({decision.reason}). " "다른 worker 가 진행 중일 가능성 — auto-force 차단됨. " "확인 후 mode='force' 로 명시 재실행하세요." ) else: _msg = decision.reason raise AppError( code="step.resume_invalid", message=_msg, status_code=409, ) # 3.5. atomic claim (Block C C1) — RERUN_SELF / FORCE_EXPLICIT / # STALE_RUNNING_RECOVERY 한정. SKIP / BLOCK 은 위에서 return / raise 처리. is_stale_steal = (decision.action == ResumeAction.STALE_RUNNING_RECOVERY) claim_ok = self._try_claim_running( allow_stale_steal=is_stale_steal, expected_started_at=decision.expected_started_at if is_stale_steal else None, expected_run_id=decision.expected_run_id if is_stale_steal else None, ) if not claim_ok: # AC-C2: claim 실패 = 다른 worker 이미 잡음 또는 stale read 후 갱신됨. raise AppError( code="step.already_running", message=( f"{self.step_id} 이 이미 다른 worker 에서 실행 중 " f"(또는 stale read 후 갱신됨). decision={decision.action.value}, " f"is_stale_steal={is_stale_steal}." ), status_code=409, ) if is_stale_steal: logger.warning( "[STALE_RUNNING] step=%s atomic steal succeeded " "(expected_started_at=%s, expected_run_id=%s, new_run_id=%s)", self.step_id, decision.expected_started_at, decision.expected_run_id, self.run_id, ) # 4. RERUN_SELF / FORCE_EXPLICIT — 부수효과 dispatch (origin 별). if decision.action == ResumeAction.RERUN_SELF and decision.origin == "artifact_missing": # D1 사고 패턴 (PID 0bb48ebf): status=completed + cp 결손/mismatch/verify 실패. # diff 로깅 — config_hash mismatch 시 어느 key 가 바뀌었는지 명시 (M3). # cp=None 인 경우 diff 의미 없음 → cp 있을 때만 진단 로그. cp = self.load_checkpoint() if cp is not None: from app.core.applicability import _diff_project_config cp_snapshot = cp.get("project_config_snapshot") try: config_diff = _diff_project_config(cp_snapshot, self.project_config) except Exception as diff_exc: # diff 계산 실패는 로깅 보조 — 본 흐름 차단 안 함. config_diff = {"_diff_error": str(diff_exc)} logger.warning( "Step %s stale detected: %s. Project config diff: %s", self.step_id, decision.reason, config_diff, ) # recovery_count++ + last_recovery_reason 영속화 (auto-rerun 가시성). new_count = self._record_recovery(decision.reason) logger.warning( "[RECOVERY] step=%s reason=%s cycle=%d", self.step_id, decision.reason, new_count, ) # 추천 2 (Audit A4 DRY 통합): record 직후 한계 검사. # 누적 recovery_count >= MAX → 자동 재시도를 차단하고 수동 진단 강제. self._check_recovery_exhausted() mode = "force" elif decision.action == ResumeAction.RERUN_SELF and decision.origin == "prior_state": # 추천 1: status='completed' 외 모든 상태는 force-like 처리. # failed = 이전 실패 / partial = 부분 성공 / stale = downstream # invalidate 결과 / pending = step_run row 만 있고 미실행. # 모두 force (cleanup + invalidate + execute) 흐름. # ('running' 은 B4 _evaluate_running_state 가 BLOCK / STALE 처리) existing = self._get_step_run(self.step_id) logger.warning( "[RECOVERY] step=%s status=%s → force-like", self.step_id, existing["status"] if existing else "?", ) mode = "force" elif decision.action == ResumeAction.RERUN_SELF and decision.origin == "contract_drift": # B5: _LEGACY_SCHEMA_BUMP_ALLOWLIST 내 step 의 schema_version mismatch. # entity_t2i 만 — 명시 force-like 진행 (사용자 의도 명시 안 해도 자동 # 재실행 허용). recovery counter 영속 (가시성). new_count = self._record_recovery(decision.reason) logger.warning( "[RECOVERY] step=%s allowlist contract_drift cycle=%d: %s", self.step_id, new_count, decision.reason, ) self._check_recovery_exhausted() mode = "force" elif decision.action == ResumeAction.FORCE_EXPLICIT: mode = "force" # RERUN_SELF + origin=None → first-run / unknown status. 현재 run() 의 # fall-through 동작 (cleanup 없이 execute) 보존 — mode='resume' 유지. # 4. 실행 경로 디스패치 (Block B B12, plan v2.1.3 / spec V5 §S2): # RERUN_SELF / FORCE_EXPLICIT 별 cleanup 정책 분리. # - RERUN_SELF: cleanup_artifacts / invalidate_downstream / clear_checkpoint # 호출 안 함 (downstream cp 보존 — auto-recovery 가 다른 step 영향 X). # - FORCE_EXPLICIT (mode='force' 또는 사용자 명시): cleanup + invalidate # + clear → finalize. if mode == "force": # mode='force' initial: 사용자 명시 force 또는 위 디스패치에서 force 격상. # FORCE_EXPLICIT (initial mode='force') 외에 RERUN_SELF + origin in # ('artifact_missing', 'prior_state', 'contract_drift') 도 mode='force' # 로 격상되어 도달 — B12 정책 (auto-recovery 도 cleanup 없이) 반영 위해 # RERUN_SELF 케이스 분기. # # Note: `mode = "force"` 격상 은 위에서 origin 별 부수효과 (record_recovery, # warning log) 처리 후 set 됐으므로, 여기서는 _execute_force / _execute_ # rerun_self 만 dispatch 하면 됨. RERUN_SELF action 인 경우 cleanup 안 함. if decision.action == ResumeAction.RERUN_SELF: return self._execute_rerun_self() # FORCE_EXPLICIT (mode='force' initial) return self._execute_force() # mode='resume' 유지 (RERUN_SELF + origin=None — first-run / unknown): # cleanup 없이 _execute_rerun_self 진행. return self._execute_rerun_self() # ── B12: rerun_self / force 실행 경로 분리 + 공통 finalize ── def _execute_rerun_self(self) -> Dict[str, Any]: """auto-recovery 경로 — 자기 step 만 rerun (Block B B12, AC-B6). force 와의 차이: - cleanup_artifacts() 호출 X - invalidate_downstream() 호출 X - clear_checkpoint() 호출 X - downstream cp 보존 (auto-recovery 가 downstream 영향 X) finalize 는 _execute_and_finalize() 공통 helper. _execute(mode='resume') 호출 — force-like 격상 X. """ return self._execute_and_finalize(execute_mode="resume") def _execute_force(self) -> Dict[str, Any]: """사용자 명시 force — cleanup + invalidate_downstream cascade (Block B B12, AC-B6). AC-B6 의 대척 경로 — force 는 의도적으로 downstream invalidate. cleanup_ artifacts 예외 시 status='failed' 영속 + raise (silent fallback 금지). finalize 는 _execute_and_finalize() 공통 helper. """ try: cleanup_report = self.cleanup_artifacts() except Exception as exc: # cleanup 실패 = DB inconsistent 위험 → 즉시 raise + failed 마킹. # silent fallback 금지 — 사용자에게 노출 + step_run에 영속화. # AC-C7: claim 후 transition. owner_lost 시 silent absorb 후 원래 exc surface. try: self._update_step_run_strict( "failed", error_message=f"cleanup_artifacts crashed: {exc}", ) except AppError as owner_exc: if owner_exc.code != "step.owner_lost": raise logger.warning( "Step %s force-cleanup owner_lost: %s (original=%s)", self.step_id, owner_exc.message, exc, ) logger.error( "Step %s cleanup_artifacts crashed: %s", self.step_id, exc, ) raise if cleanup_report.deleted_db_rows or cleanup_report.deleted_files: logger.warning( "[CLEANUP] step=%s rows=%d files=%d targets=%s", self.step_id, cleanup_report.deleted_db_rows, cleanup_report.deleted_files, cleanup_report.targets, ) self.invalidate_downstream() self.clear_checkpoint() return self._execute_and_finalize(execute_mode="force") def _execute_and_finalize(self, *, execute_mode: str = "resume") -> Dict[str, Any]: """rerun_self / force 두 경로 공통 finalize (Block B B12, plan v2.1.3 §S2). 7 단계 (V5 S2 + P3): (1) _update_step_run("running") + opik context 설정 (2) _execute(mode=execute_mode) + _last_execute_result attribute (3) final_status 계산 (failed/partial/completed) (4) exit verify (final_status='completed' 시) — `step.verify_crashed` AppError raise 시 status='failed' 영속 후 re-raise (running leak 차단). 그 외 verify is_complete=False → final_status='partial' 격상. (5) _update_step_run(final_status, ...) (6) save_checkpoint({"status": final_status, **result}) + recovery counter reset (final_status='completed' 시) (7) modifies_checkpoints cascade policy (1-pass default) 예외 처리: _execute / finalize 단계의 unexpected exception → status='failed' 영속 + re-raise. status='running' leak 차단 (다음 resume 에서 stale 분기 회피). """ from app.modules.llm.llm_client import set_opik_context # (1) running mark + opik. claim 직후 호출 — strict (AC-C7). # _try_claim_running 이 이미 status='running' 으로 row 잡았으므로 본 호출은 # progress fields reset / opik metadata 동기화 의도. owner check 실패 시 # (review I2: claim ↔ 본 호출 사이 race steal) expensive _execute 진입 전에 # 즉시 step.owner_lost AppError 로 short-circuit. self._update_step_run_strict("running") logger.info( "Step %s started (run_id=%s, model=%s)", self.step_id, self.run_id, self._resolve_model(), ) set_opik_context(self.build_opik_metadata()) # V2.1.1 patch: inner exit verify 가 specific message 로 status='failed' 기록 # 한 경우 outer except 의 generic message overwrite 차단. failed_recorded = False try: # (2) _execute result = self._execute(mode=execute_mode) # Group 1 #3: cp fallback 외 source — _last_execute_result attribute. self._last_execute_result = result completed = result.get("completed_count", 1) total = result.get("applicable_count", 1) failed = result.get("failed_count", 0) # (3) final_status final_status = ( "completed" if failed == 0 else ("partial" if completed > 0 else "failed") ) # (4) exit verify (B7 + B12: verify_crashed 시 status='failed' 영속). if final_status == "completed": try: exit_report = self._safe_verify_completion() except AppError as verify_exc: logger.error( "[VERIFY-EXIT] step=%s exit verify crashed: %s — %s", self.step_id, getattr(verify_exc, "code", "?"), verify_exc.message, ) # claim 후 transition — owner-aware (AC-C7). owner_lost 시 # silent absorb (race lost) 후 원래 verify_exc 보존. try: self._update_step_run_strict( "failed", completed_count=completed, applicable_count=total, failed_count=failed, error_message=( f"exit verify crashed: " f"{getattr(verify_exc, 'code', '?')} — {verify_exc.message}" ), ) except AppError as owner_exc: if owner_exc.code != "step.owner_lost": raise logger.warning( "[VERIFY-EXIT] step=%s owner_lost on failed-mark: %s", self.step_id, owner_exc.message, ) failed_recorded = True raise if not exit_report.is_complete: logger.warning( "[VERIFY-EXIT] step=%s failed → partial: %s", self.step_id, exit_report.missing, ) final_status = "partial" # (5) status transition — claim 후 owner-aware (AC-C7). self._update_step_run_strict( final_status, completed_count=completed, applicable_count=total, failed_count=failed, result_summary=json.dumps( {k: v for k, v in result.items() if k != "data" and not isinstance(v, bytes)}, ensure_ascii=False, default=str, )[:2000], ) # (6) save_checkpoint + recovery counter reset self.save_checkpoint({"status": final_status, **result}) if final_status == "completed": self._reset_recovery_counter() # (7) modifies_checkpoints cascade (editorial step 지원) mods = self.manifest.get("modifies_checkpoints") or [] cascade_on = self.manifest.get("invalidate_downstream_on_edit", False) if final_status in ("completed", "partial") and mods: if cascade_on: for target_sid in mods: self.invalidate_downstream( target_step_id=target_sid, delete_checkpoints=False, ) else: logger.debug( "Step %s: editorial cascade skipped (policy=1-pass, targets=%s)", self.step_id, mods, ) logger.info( "Step %s %s (completed=%d/%d, failed=%d)", self.step_id, final_status, completed, total, failed, ) return {"status": final_status, "result": result} except Exception as exc: import traceback # V2.1.1 patch: failed_recorded 시 specific exit verify message 보존. # AC-C7: claim 후 transition 이므로 owner-aware. owner_lost 시 silent # absorb (race lost — 원래 exc 가 우선 surface). if not failed_recorded: try: self._update_step_run_strict("failed", error_message=str(exc)) except AppError as owner_exc: if owner_exc.code != "step.owner_lost": raise logger.warning( "Step %s exception path owner_lost: %s (original=%s)", self.step_id, owner_exc.message, exc, ) logger.error( "Step %s _execute_and_finalize crashed: %s\n%s", self.step_id, exc, traceback.format_exc(), ) raise finally: set_opik_context(None) # ── 무결성 검증 (Phase Resume Integrity) ── def verify_completion(self) -> "CompletionReport": """산출물 무결성 검증. 베이스는 항상 complete — override 안 한 step은 검증 없음. Resume entry + execute exit 양방향에서 호출됨. is_complete=False 반환 시: - entry: mode='force' 격상 (auto-rerun) - exit: status='partial' 마킹 Override 시 자기 step 산출물(DB row + 파일 stat)만 검증한다. cross-step dep verify는 후속 spec. """ from app.core.integrity_report import CompletionReport return CompletionReport(is_complete=True, missing=[], severity="clean", metadata={}) def cleanup_artifacts(self) -> "CleanupReport": """force/auto-rerun 시 stale artifact 정리. 베이스는 noop. 호출 site: run() force 분기 (Task 12 구현 예정)에서 _execute() 직전 1회 호출. resume entry verify가 is_complete=False 반환 → mode='force' 격상 → cleanup → execute 흐름. Override 작성 기준 (매우 보수적): - 전체 force = delete-and-rebuild가 항상 안전한 step만 - 부분 재생성을 endpoint로 지원하는 step은 절대 override 금지 - 이미지 step은 거의 모두 default noop 유지 (사용자 caveat) """ from app.core.integrity_report import CleanupReport return CleanupReport(deleted_db_rows=0, deleted_files=0, targets=[], skipped=[]) # ── recovery counter (auto-rerun 가시성, spec §4.3 추천 2) ── def _check_recovery_exhausted(self) -> None: """누적 recovery_count >= MAX_RECOVERY_ATTEMPTS면 step.recovery_exhausted raise. Audit A4 (DRY): 기존 dual gate(resume 분기 record 전 + force 분기 진입 직전)를 한 곳으로 통합. 호출 site는 `_record_recovery` **직후** — post-increment된 새 카운트로 즉시 검사하여 force 분기 진입 자체를 차단한다. message에는 module 상수 MAX_RECOVERY_ATTEMPTS를 명시 — silent recovery로 묻히지 않게 사용자에게 한계값을 노출하고, 회귀 테스트의 "3" 포함 검사도 보존한다(누적 카운트와 한계가 모두 표시됨). """ count = self._get_recovery_count() if count >= MAX_RECOVERY_ATTEMPTS: raise AppError( code="step.recovery_exhausted", message=( f"{self.step_id} auto-recovery 한계({MAX_RECOVERY_ATTEMPTS}회) 도달 — " f"누적 {count}회 실패. 마지막 사유: {self._get_last_recovery_reason()}. " "수동 진단 필요." ), status_code=409, ) def _record_recovery(self, reason: str) -> int: """recovery_count++ + last_recovery_reason 적재. 새 카운터값 반환. 호출 site: entry verify 실패 시. mode='force' 격상 직전. """ now = self._now() self.db.execute(text(""" UPDATE step_run SET recovery_count = recovery_count + 1, last_recovery_reason = :reason, updated_at = :now WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid """), {"reason": reason[:2000], "pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "now": now}) self.db.commit() return self._get_recovery_count() def _reset_recovery_counter(self) -> None: """status='completed' 정상 진행 시 카운터 reset (다음 sweep 깨끗한 상태).""" now = self._now() self.db.execute(text(""" UPDATE step_run SET recovery_count = 0, last_recovery_reason = NULL, updated_at = :now WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid """), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "now": now}) self.db.commit() def _get_recovery_count(self) -> int: row = self.db.execute(text( "SELECT recovery_count FROM step_run " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid" ), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id}).fetchone() return row[0] if row else 0 def _get_last_recovery_reason(self) -> str: row = self.db.execute(text( "SELECT last_recovery_reason FROM step_run " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid" ), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id}).fetchone() return (row[0] if row and row[0] else "(없음)") @staticmethod def _mask_sensitive_keys(config) -> dict: """project_config 저장 전 민감 키 마스킹. api_key/password/secret/token/credential 패턴. nested dict는 재귀. None → 빈 dict. D2 fix — save_checkpoint이 project_config_snapshot 기록 시 사용. """ SENSITIVE = ("api_key", "password", "secret", "token", "credential") if not config: return {} masked = {} for k, v in config.items(): if any(s in str(k).lower() for s in SENSITIVE): masked[k] = "***" elif isinstance(v, dict): masked[k] = StepRunner._mask_sensitive_keys(v) else: masked[k] = v return masked def _safe_verify_completion(self) -> "CompletionReport": """verify_completion 실행 + 예외 분류 (Block B B7, plan v2.1.3 / spec V5 §4.3). - AppError(code='step.verify_crashed') → propagate (B8 verify_completion 이 card recompute 등에서 미리 격상한 케이스 — 이중 wrap 금지). - 그 외 AppError (loader contract / step.contract_violation 등) → CompletionReport(origin='contract_drift', is_complete=False). caller (run() / ResumeDecision helper) 가 BLOCK 처리 (B11 후속). - 그 외 Exception (KeyError / ValueError / AttributeError 등) → AppError(step.verify_crashed) raise (자동 force 금지, fail-fast). silent recovery 사고 패턴 차단. """ from app.core.integrity_report import CompletionReport try: return self.verify_completion() except AppError as exc: # 이미 verify_crashed 격상된 경우 → 원형 propagate if getattr(exc, "code", None) == "step.verify_crashed": raise logger.warning( "verify_completion AppError for %s: %s (code=%s)", self.step_id, exc.message, getattr(exc, "code", "?"), ) return CompletionReport( is_complete=False, missing=[f"contract: {exc.message}"], severity="missing", metadata={"app_error_code": getattr(exc, "code", None)}, origin="contract_drift", ) except Exception as exc: # unexpected exception — fail-fast (자동 force 금지) logger.error( "verify_completion crashed for %s: %s", self.step_id, exc, exc_info=True, ) raise AppError( code="step.verify_crashed", message=f"verify_completion crashed: {type(exc).__name__}: {exc}", ) from exc def _check_cp_mismatch(self, cp: Dict[str, Any]) -> Optional[str]: """cp의 schema_version + config_hash와 현재 manifest/project_config 비교. Returns: mismatch reason str (None이면 일치). run() resume 분기의 P0-3 mismatch 검증을 헬퍼로 이관. legacy 체크포인트(cp_schema/cp_hash 누락)는 None 반환 — 호환 보장. config_hash 비교 (Codex iter root cause): save_checkpoint (line 349-350) 는 step 이 _execute() 에서 명시 반환한 'config_hash' 가 있으면 그것을 보존하고, 없으면 compute_config_hash( project_config) 로 fallback. 따라서 mismatch check 도 step 의 자체 hash method (`_config_hash()`) 가 있으면 그것으로 계산해야 일관. 이전 결함: _execute 가 SHA256(prompt_version+model+...) hash 저장 → check 는 MD5(project_config) hash 비교 → 영구 false-positive mismatch. """ current_schema = self.manifest.get("schema_version", 1) cp_schema = cp.get("schema_version") cp_hash = cp.get("config_hash") if cp_schema is not None and cp_schema != current_schema: return f"schema_version mismatch: 체크포인트={cp_schema}, 현재={current_schema}" if cp_hash is not None: # step-local hash override (background_classify, background_master_plan # 등) 가 있으면 같은 method 로 비교. 없으면 project_config-only fallback. local_hash_fn = getattr(self, "_config_hash", None) if callable(local_hash_fn): # (era R3 잔여 HIGH) fail-closed — _config_hash() 예외를 # project_config hash 로 조용히 대체하면 legacy CP 와의 # 우연 일치=stale SKIP(진짜 drift 은폐), 불일치=drift 오진 # 이 된다. 저장 경로(image_steps._config_hash 호출)는 # 무보호라 크게 죽으므로 비교 게이트도 대칭으로 전파한다 # — 해시 결함은 계약 판정 불능이지 drift 가 아니다. current_hash = local_hash_fn() hash_kind = "step-local" else: current_hash = compute_config_hash(self.project_config) hash_kind = "project_config" if cp_hash != current_hash: return ( f"config_hash mismatch ({hash_kind}): " f"cp={cp_hash}, current={current_hash}" ) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: """서브클래스에서 구현. 실제 LLM 호출 수행. Returns: { "completed_count": N, "applicable_count": N, "failed_count": N, "data": {...}, # 단계별 결과 데이터 } """ raise NotImplementedError(f"Step {self.step_id} must implement _execute()") # ── 유틸 ── def _resolve_model(self) -> str: """현재 단계의 모델 별칭 반환.""" from app.modules.llm.llm_client import _resolve_model return _resolve_model(self.step_id, self.project_config) def build_opik_metadata( self, extra_tags: Optional[List[str]] = None, extra_metadata: Optional[Dict[str, Any]] = None, ) -> Dict: """Opik metadata 구성 — thread 그룹핑 + 프로젝트/에피소드 정보. extra_tags: Opik 태그에 추가할 낮은 카디널리티 문자열만 사용 (예: "retry", "gpt_fallback"). 동적 ID(scene_index, shot_index 등)는 태그가 아니라 extra_metadata에 넣어야 Opik 인덱스 카디널리티 폭발을 방지한다. extra_metadata: 메타데이터 dict에 merge. 필터링은 metadata 기반으로 가능. """ # 신원 두 칸은 항상 싣는다. litellm 을 우회하는 호출(검색·이미지)은 # `ambient_call_meta()` 로 여기서 신원을 읽어 DB·Opik 에 남긴다 — # 호출자마다 ID 를 넘기게 만들면 배선을 빠뜨린 자리가 조용히 기록 없이 # 돌기 때문이다. meta: Dict[str, Any] = { "project_id": self.project_id, "episode_id": self.episode_id, } ctx = self.opik_context if ctx.get("run_tag"): meta["trace_name"] = f"{ctx.get('project_name', '')} > {ctx.get('episode_title', '')} > {self.step_id}" meta["session_id"] = ctx["run_tag"] # Opik thread grouping tags = [self.step_id] if ctx.get("project_name"): meta["project_name_tag"] = ctx["project_name"] tags.append(ctx["project_name"]) if ctx.get("episode_title"): meta["episode_title_tag"] = ctx["episode_title"] tags.append(ctx["episode_title"]) if extra_tags: tags.extend(extra_tags) meta["tags"] = tags if extra_metadata: for k, v in extra_metadata.items(): if k not in meta: meta[k] = v return meta @staticmethod def _now() -> str: return datetime.now(timezone.utc).isoformat()