"""StepRunner 베이스 클래스 — 모든 파이프라인 단계의 실행 프레임워크. 각 단계는 이 클래스를 상속하여 _execute()만 구현하면 됨. 게이트, 체크포인트, 무효화, Opik 추적을 자동 처리. """ import hashlib import json import logging import os import re import uuid from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone from enum import Enum from pathlib import Path from typing import Any, Dict, List, Optional, Tuple, 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 from app.core.step_lock import ( LockKey, LockOwner, OwnerVerdict, REGISTRY as LOCK_REGISTRY, judge_owner, process_identity, ) # 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" @contextmanager def _step_trace_scope(runner): """스텝 하나를 Opik trace 하나로 연다. 설정이 꺼져 있으면 아무것도 안 한다(바이트 동일). 여는 데 실패해도 본 스텝은 그대로 돈다 — `open_trace` 가 None 을 줄 뿐이다. """ from app.modules.llm.opik_trace import open_trace meta = dict(runner.build_opik_metadata() or {}) tags = list(meta.pop("tags", []) or []) thread_id = meta.get("thread_id") with open_trace( name=f"step:{runner.step_id}", tags=tags, metadata=meta, thread_id=thread_id, ): yield @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 # ★claim **전** step_run 행의 상태 (Codex 2026-09-03 07:35 · 실측 f7cc45c576c0): 판정 뒤 claim 이 행을 running 으로 # 바꾸므로, 로그·감사는 이 값만 믿는다. 행이 없으면 None. prior_status: Optional[str] = None prior_run_id: Optional[str] = None prior_updated_at: Optional[str] = None expected_started_at: Optional[str] = None expected_run_id: Optional[str] = None #: ★값에 따라 지문에서 빼는 칸. **key 째로 빼지 않는다.** #: #: `grounding_mode` 는 `legacy`/`shadow_plan` 일 때만 뺀다. ★이유가 서로 **다르다**: #: #: · `legacy` — **키가 없는 것과 같은 동작**이라서 뺀다. legacy 가 무료라는 뜻이 #: **아니다** — legacy 는 기존 장소 시대 조사(`ERA_RESEARCH_ENABLED=true`)와 #: 정상 이미지 파이프라인을 그대로 돈다. #: · `shadow_plan` — **관측 전용**이라서 뺀다. 검색도 이미지 생성도 안 하고 #: production 산출을 한 바이트도 안 바꾸는데, dict 전체를 해시하면 그 스위치를 #: 켠 것만으로 detail·t2i·reference·scene 이 다시 구워진다 (계약 §12). #: #: **`v2` 는 반드시 접는다** — 안 접으면 legacy 완료 CP 가 v2 진입에서 재사용된다. _VALUE_NORMALIZED_CONFIG_KEYS = {"grounding_mode"} 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 타입은 클래스명만 기록. ★**`grounding_mode` 는 값에 따라 접거나 뺀다** (`_VALUE_NORMALIZED_CONFIG_KEYS`). 실측: 이 함수는 dict 전체를 해시하므로 그 키를 넣기만 해도 지문이 움직였고 `shadow_plan` 으로 바꾸면 또 움직였다 — 검색도 산출도 안 바꾸는 관측 스위치가 하류를 통째로 다시 굽게 만든다 (계약 §12). ★**그렇다고 key 째로 빼면 안 된다.** `v2` 까지 숨겨져 **legacy 완료 CP 가 v2 진입에서 그대로 재사용된다.** 「revision/content hash 가 대신 잡는다」는 전제는 아직 구현이 없다 (그 값을 만드는 production 코드 0건). Returns 16자 md5 prefix. """ from app.core.grounding_mode import fingerprint_value, resolve_grounding_mode payload = dict(project_config or {}) # ★**실효 모드**로 접는다 — dict 에 키가 있는지로 보면 안 된다 (Codex BLOCK). # `resolve_grounding_mode` 는 키가 없거나 None 이면 `GROUNDING_MODE` 환경변수로 # 떨어진다. 키만 보면 **ENV=v2 + 키 없음** 일 때 실제 실행은 v2 인데 지문은 # bare/legacy 와 같아져서, 막으려던 legacy CP 재사용이 그대로 돌아온다. # (재현: ENV=v2 로 두면 키 없는 지문이 legacy 지문과 같았다) effective = resolve_grounding_mode(payload) kept = fingerprint_value(effective) if kept is None: payload.pop("grounding_mode", None) else: payload["grounding_mode"] = kept canonical = json.dumps( payload, 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: # ★도는 중인 하류는 status 만 바꾸면 안 된다 (2026-08-26 Codex # 2차 재리뷰 BLOCK-3). 그 worker 는 자기 토큰이 그대로라 아무것도 # 모른 채 계속 돌고 유료 호출을 이어 간다. 토큰을 같이 버리면 # 다음 안전 지점(`checkpoint_gate`)에서 바로 멈춘다. # # ★`shot_selection_service` 에만 넣고 여기 중앙 경로를 빼 두면 # 같은 좀비가 그대로 남는다 — 단일 스텝 실행끼리는 에피소드 # 자리 밖에서 겹칠 수 있다. res = self.db.execute(text( "UPDATE step_run SET status = 'stale', updated_at = :now, " " run_id = CASE WHEN status = 'running' THEN :revoked " " ELSE run_id END " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid " "AND status NOT IN ('pending', 'stale') " "RETURNING run_id" ), {"pid": self.project_id, "eid": self.episode_id, "sid": sid, "now": now, "revoked": f"invalidated-{uuid.uuid4()}"}) _row = res.fetchone() if _row and str(_row[0]).startswith("invalidated-"): logger.warning( "invalidate_downstream: 도는 중인 하류 %s 를 무효화했다 — " "그 주행은 다음 안전 지점에서 멈춘다 (ref=%s)", sid, ref, ) 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, owner_host, owner_pid, owner_boot_id, heartbeat_at, cancel_requested_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, # 락 소유자 신원 (010_step_lock_ownership) — 죽음을 경과 시간으로 # 추측하지 않고 확인하기 위한 재료. 구 행은 전부 None. "owner_host": row.owner_host, "owner_pid": row.owner_pid, "owner_boot_id": row.owner_boot_id, "heartbeat_at": row.heartbeat_at, "cancel_requested_at": row.cancel_requested_at, } @property def lock_key(self) -> "LockKey": """이 러너가 잡는 락의 신원 — 등록부 조회 키.""" return (self.project_id, self.episode_id, self.step_id) # ── 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: """★claim 전 행의 상태를 판정에 싣는다 — 판정 본체는 `_evaluate_resume_decision_inner`.""" import dataclasses existing = self._get_step_run(self.step_id) decision = self._evaluate_resume_decision_inner(mode) if existing: decision = dataclasses.replace( decision, prior_status=str(existing.get("status") or "") or None, prior_run_id=(str(existing.get("run_id")) if existing.get("run_id") else None), prior_updated_at=(str(existing.get("updated_at")) if existing.get("updated_at") else None)) return decision def _evaluate_resume_decision_inner(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) # 'cancelled' = 운영자가 세운 것. 재개하면 그냥 다시 하면 된다 — # 실패와 같은 취급(prior_state)이되 사유는 구별해 남긴다. # ★주행 단위 정지 표가 아직 걸려 있으면 스텝을 시작해도 첫 안전 지점에서 # 다시 선다. 표를 내리는 것은 `POST .../cancel/clear` 다. if status in ("failed", "partial", "stale", "pending", "cancelled"): 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 평가 — **죽음을 추측하지 않고 확인한다.** 2026-08-26 개편. 이전에는 `started_at` 경과 시간만으로 판정해서, 프로세스를 강제 종료하면 `step_running_timeout_seconds`(기본 3600초)가 지날 때까지 아무것도 못 했다 (새벽 실측 손실 40분). 지금은 `step_lock.judge_owner()` 가 소유자 신원으로 먼저 판정하고, 신원을 못 믿는 자리에서만 시간으로 떨어진다: - `OwnerVerdict.DEAD` → STALE_RUNNING_RECOVERY (바로 풀기) - `OwnerVerdict.ALIVE` → BLOCK (진짜로 일하는 중) - `OwnerVerdict.UNKNOWN` → 기존 경과 시간 경로 (신원 칸이 없는 구 행) ★`started_at` 이 없거나 깨진 것은 더 이상 무조건 BLOCK 이 아니다 — 신원이 죽음을 말하면 그것이 우선한다. 시간은 신원이 침묵할 때만 쓰는 마지막 그물이다. 본 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") # ── 1. 소유자 신원으로 판정 (추측 아님) ── owner = LockOwner.from_row(existing) verdict, verdict_reason = judge_owner( owner, key=self.lock_key, lease_seconds=settings.step_lock_lease_seconds, allow_heartbeat_steal=settings.step_lock_heartbeat_steal_enabled, ) if verdict is OwnerVerdict.DEAD: return ResumeDecision( action=ResumeAction.STALE_RUNNING_RECOVERY, reason=f"소유자가 죽었다 — {verdict_reason}", expected_started_at=started_at_raw, expected_run_id=run_id_raw, ) if verdict is OwnerVerdict.ALIVE: return ResumeDecision( action=ResumeAction.BLOCK, reason=f"소유자가 살아있다 — {verdict_reason}", origin="running_healthy", ) # ── 2. UNKNOWN — 확정 못 했다. 무엇이 다음인가는 **신원이 있느냐**로 갈린다. ── # # ★신원 칸이 하나라도 있는 행은 경과 시간으로 안 뺏는다. 뺏으면 # 「확정 사망만 자동 푼다」는 계약이 3600초 뒤에 무너진다 — 하트비트가 # 멈춘 것을 죽음으로 안 읽기로 해 놓고, 시간이 지나면 결국 같은 일을 # 하는 셈이다. 이런 행은 운영자가 `GET .../locks` 로 보고 `release` 로 # 명시 해제한다. 그 길이 이제 있다. # # ★경과 시간 경로는 **신원이 통째로 없는 옛 행**에만 남긴다. 그 행들에 # 대해서는 오늘까지의 동작 그대로다 (새 위험 0). has_identity = any(( existing.get("owner_boot_id"), existing.get("owner_host"), existing.get("heartbeat_at"), )) if has_identity: return ResumeDecision( action=ResumeAction.BLOCK, reason=( f"소유자 생사를 확정 못 했다 — {verdict_reason}. " "GET .../steps/locks 로 확인하고, 죽은 것이 맞으면 " "POST .../steps/locks/{step_id}/release 로 풀어라." ), origin="running_healthy", ) if started_at_raw is None: return ResumeDecision( action=ResumeAction.BLOCK, reason=( f"running with started_at=None (manual investigation required); " f"{verdict_reason}" ), 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}; " f"{verdict_reason}" ), 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); " f"{verdict_reason}" ), origin="running_healthy", ) return ResumeDecision( action=ResumeAction.STALE_RUNNING_RECOVERY, reason=( f"running stale (elapsed={elapsed:.0f}s >= timeout={timeout_seconds}s); " f"{verdict_reason}" ), 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", ) #: 마무리가 터미널 상태를 적은 **뒤에** 실패를 되돌려 적을 때 허용하는 #: 출발 상태. 밖에서 'stale' 로 갈아 끼운 줄은 여전히 못 덮는다. _TERMINAL_ROLLBACK_FROM = ("running", "completed", "partial") def _update_step_run( self, status: str, *, require_owner: bool = False, expected_status: Tuple[str, ...] = ("running",), 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 판정). `expected_status` — **이 전이가 어느 상태에서 출발해야 하는가.** ★상태 조건을 `'running'` 하나로 못박았다가 성공 주행을 망가뜨렸다 (2026-08-26 Codex PR#4 리뷰 BLOCK-1·2). fan-out 스텝은 마지막 조각을 끝내며 진행률을 적는데, 그 진행률이 DB 를 곧바로 'completed' 로 바꿔 놓으면 뒤이은 마무리가 「내 줄이 아니다」로 읽혀 **성공이 예외로 끝났다.** 마무리가 completed 를 적은 뒤 체크포인트 저장이 실패해 failed 로 되돌리려 할 때도 같은 자리에서 막혀 **DB 만 성공으로 남았다.** 그래서 조건을 없애지 않고 **전이마다 출발 상태를 밝히게** 했다. 밖에서 status 만 갈아 끼운 줄을 못 덮는다는 원래 성질은 그대로다. """ now = self._now() started_at = now if status == "running" else None completed_at = now if status in ("completed", "not_applicable") else None # ★`run_id` 만으로는 모자란다 (2026-08-26 Codex 재리뷰 BLOCK-4). # 토큰을 바꾸지 않고 **status 만** 밖에서 갈아 끼우는 경로가 있다 # (`shot_selection_service` 의 downstream stale 표시 등). 그런 # 줄을 worker 가 completed 로 덮으면 그 표시가 조용히 사라진다. # # ★정지 요청은 status 를 안 바꾼다(`cancel_requested_at` 만 세운다) # — 그래서 정지 뒤 'cancelled' 로 적는 길은 이 조건에 안 걸린다. _exp_keys: List[str] = [] owner_clause = "" if require_owner: _exp_keys = [f"exp{i}" for i in range(len(expected_status))] _in = ", ".join(f":{k}" for k in _exp_keys) owner_clause = ( f"WHERE step_run.run_id = :rid AND step_run.status IN ({_in})" ) 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, **{k: v for k, v in zip(_exp_keys, expected_status)}, }) # 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, *, expected_status: Tuple[str, ...] = ("running",), 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, expected_status=expected_status, 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()) now = self._now() owner_host, owner_pid, owner_boot_id = process_identity() # 락에 **소유자 신원**을 함께 적는다 (010_step_lock_ownership). 이 칸이 # 있어야 다음 사람이 죽음을 추측 대신 확인할 수 있다. # `cancel_requested_at` 은 새 claim 마다 지운다 — 지난 주행의 정지 # 요청이 새 주행을 죽이면 안 된다. sql = text(""" INSERT INTO step_run ( id, project_id, episode_id, step_id, run_id, status, started_at, created_at, updated_at, owner_host, owner_pid, owner_boot_id, heartbeat_at, cancel_requested_at ) VALUES ( :new_step_run_id, :pid, :eid, :sid, :run_id, 'running', :now, :now, :now, :owner_host, :owner_pid, :owner_boot_id, CURRENT_TIMESTAMP, NULL ) 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, owner_host = EXCLUDED.owner_host, owner_pid = EXCLUDED.owner_pid, owner_boot_id = EXCLUDED.owner_boot_id, heartbeat_at = EXCLUDED.heartbeat_at, cancel_requested_at = NULL, 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": now, "owner_host": owner_host, "owner_pid": owner_pid, "owner_boot_id": owner_boot_id, "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 checkpoint_gate(self, *, where: str = "", strict: bool = False) -> None: """안전 지점 검사 — **아직 내가 주인인가, 멈추라는 말이 왔나.** 팬아웃 스텝이 한 단위(샷 하나 등)를 끝낼 때, 그리고 **돈을 쓰기 직전**에 부른다. 세 가지를 막는다: 1. **정지 요청** (`cancel_requested_at` 또는 `run_cancel_request`). 운영자가 `POST .../cancel` 을 불렀다. 2. **락 풀기** (`status != 'running'`). `release` 가 이 락을 풀기해 `failed` 로 적었다. `run_id` 는 그대로일 수 있으므로 **상태도 봐야** 한다 — 안 보면 풀린 뒤에도 옛 worker 가 그냥 통과한다. 3. **좀비 쓰기** (`run_id` 불일치). 다른 worker 가 락을 가져갔다. 소유권 검사는 지금까지 **마지막 상태 갱신에만** 있어서, 그 전에 일어나는 DB commit·파일 저장은 못 막았다. Args: where: 어디서 불렀는지 (오류 문구에 들어간다). strict: DB 를 못 읽었을 때 **멈출지**. 기본 False. ★`strict` 를 가르는 이유 — 검사 조회가 실패했을 때 그냥 통과시키면 DB 장애 동안 좀비가 계속 쓴다. 반대로 항상 멈추면 잠깐의 접속 장애로 정상 주행이 죽는다. 그래서 **일반 안전 지점은 통과**(fail-open), **돈을 쓰거나 결과물을 확정하기 직전은 멈춤**(fail-closed)으로 가른다. ★**전용 세션**을 쓴다. 이 검사은 팬아웃 worker 스레드에서도 불리는데, SQLAlchemy 세션은 스레드 안전하지 않다. 게다가 PostgreSQL 은 실패한 문장 하나로 트랜잭션이 abort 되므로, 작업용 세션으로 조회했다가 실패하면 그 뒤 **그 worker 의 모든 쓰기가 죽는다.** Raises: AppError(step.cancelled): 정지 요청이 왔다. AppError(step.owner_lost): 락을 빼앗겼거나 풀렸다. AppError(step.gate_unreadable): strict 인데 검사을 못 읽었다. """ from app.core.database import SessionLocal from app.core.run_control import read_cancel_state session = SessionLocal() try: row = session.execute(text(""" SELECT run_id, status, cancel_requested_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": self.step_id, }).fetchone() # ★`strict` 를 그대로 넘긴다. 안 넘기면 이 조회가 DB 오류를 삼켜 # 「정지 없음」을 돌려주고, 검사은 fail-closed 라고 믿으면서 통과시킨다 # — 확인했다고 말하면서 실은 못 본 것이다. run_cancel = read_cancel_state( session, self.project_id, self.episode_id, strict=strict ) except Exception as exc: session.rollback() if strict: raise AppError( code="step.gate_unreadable", message=( f"{self.step_id} 소유권 검사을 못 읽었다 " f"({where or '안전 지점'}): {exc}. " "돈을 쓰기 전이라 멈춘다." ), status_code=503, ) from exc logger.warning( "checkpoint_gate 읽기 실패 (통과시킨다) step=%s where=%s: %s", self.step_id, where, exc, ) return finally: session.close() if row is None: # 행이 사라졌다 — 판단 근거가 없다. 돈 쓰기 직전이면 멈춘다. if strict: raise AppError( code="step.owner_lost", message=( f"{self.step_id} step_run 행이 없다 " f"({where or '안전 지점'}) — 돈을 쓰기 전이라 멈춘다." ), status_code=409, ) return if row.run_id != self.run_id: raise AppError( code="step.owner_lost", message=( f"{self.step_id} 락을 빼앗겼다 (내 run_id={self.run_id}, " f"DB run_id={row.run_id}) — {where or '안전 지점'}에서 멈춘다." ), status_code=409, ) if row.status != "running": # `release` 가 풀었다. run_id 는 그대로일 수 있으므로 여기서만 잡힌다. raise AppError( code="step.owner_lost", message=( f"{self.step_id} 락이 풀렸다 (status={row.status}) — " f"{where or '안전 지점'}에서 멈춘다." ), status_code=409, ) if row.cancel_requested_at: raise AppError( code="step.cancelled", message=( f"{self.step_id} 정지 요청 ({row.cancel_requested_at}) — " f"{where or '안전 지점'}에서 멈춘다." ), status_code=409, ) # 주행 단위 정지 — 이 스텝에 표가 없어도 에피소드 전체가 멈춤 요청을 # 받았을 수 있다 (`run_cancel_request`). if run_cancel.requested: raise AppError( code="step.cancelled", message=( f"{self.step_id} 주행 {run_cancel.describe()} — " f"{where or '안전 지점'}에서 멈춘다." ), status_code=409, ) 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. """ # ★진행률은 **끝까지 'running'** 이다 (2026-08-26 Codex PR#4 BLOCK-1). # 마지막 조각에서 'completed' 를 적어 버리면, 뒤이은 마무리가 # 「출발 상태가 running 이 아니다」로 읽혀 **성공 주행이 예외로 # 끝난다.** 터미널 전이는 마무리 한 곳만 소유한다. self._update_step_run_strict( "running", completed_count=completed, applicable_count=total, failed_count=failed, ) # ── 실행 ── def _preflight_grounding_mode(self) -> None: """`grounding_mode` 오타를 **첫 지출 전에** 세운다 (fail-closed). ``resolve_grounding_mode`` 가 enum 밖 값에 ``AppError`` 를 던진다. 여기서 부르는 것은 「값을 쓰려고」가 아니라 **「틀린 값이면 아무것도 시작하지 않으려고」**다. """ from app.core.grounding_mode import resolve_grounding_mode resolve_grounding_mode(self.project_config) 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-a. grounding mode preflight — ★**어떤 지출보다도 먼저** 오타를 세운다. # # ★`compute_config_hash` 안의 방어만으로는 늦다 (Codex BLOCK). 기존 완료 # CP 가 있으면 `_check_cp_mismatch` 가 hash 를 먼저 계산해 실행 전에 서지만, # **fresh/no-CP 경로**에서는 `_evaluate_resume_decision` 이 step_run 행이 # 없다는 이유로 곧바로 RERUN_SELF 를 돌려주고, 그 갈래의 첫 hash 계산은 # `save_checkpoint()` 다 — 즉 **`_execute` 와 유료 호출이 끝난 뒤**다. # 새 에피소드의 오타 하나가 돈을 다 쓰고 나서야 실패한다. # # ★`validate_mode` 에 넣지 않는다 — `SceneImagePipelineStep` 이 그것을 # override 하면서 `super()` 를 부르지 않아 **정작 제일 비싼 경로가 빠진다.** self._preflight_grounding_mode() # 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 처리. # # ★신원을 **등록보다 먼저** 확정한다. `process_identity()` 는 PID 가 # 바뀐 것을 보면 등록부를 비우는데(fork 자식이 부모 등록부를 물려받는 # 것을 막으려고), 그 첫 호출이 `_try_claim_running` 안에 있으면 # **방금 등록한 자기 항목이 지워진다.** 그러면 DB 에는 새 boot_id 로 # 잡혀 있는데 등록부는 비어 있어, 자기 자신을 죽었다고 읽고 하트비트도 # 안 뛴다. 여기서 한 번 불러 그 창을 없앤다. process_identity() # ★등록부 등록은 **claim 보다 먼저**다. claim 성공 직후·등록 직전의 # 창에서 다른 스레드가 `judge_owner` 를 부르면, DB 는 이 프로세스가 # 소유자라 말하는데 등록부에는 없어 **살아있는 락을 죽었다고** 읽는다. # 먼저 등록하고, claim 이 실패하면 되돌린다. LOCK_REGISTRY.register(self.lock_key, self.run_id) try: return self._claim_and_run(decision, mode) finally: LOCK_REGISTRY.unregister(self.lock_key, self.run_id) def _claim_and_run(self, decision: ResumeDecision, mode: str) -> Dict[str, Any]: """claim 이후의 실행 — 등록부 등록이 끝난 상태에서만 불린다. `run()` 에서 떼어낸 이유는 하나다: 락 등록부의 해제를 `finally` 로 보장하려면 claim 이후 전체가 한 블록이어야 하는데, 그러면 100줄 넘는 본문이 통째로 한 칸 들어가 diff 가 안 읽힌다. """ 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 처리) # ★claim 뒤 상태(늘 running)가 아니라 **판정 사유에 담긴 claim 전 상태**를 적는다 — 실측 f7cc45c576c0 (2026-09-03): # plain resume 가 scene_detail 을 다시 샀는데 로그가 「status=running」만 찍어 무엇이 prior 였는지 안 남았다. logger.warning( "[RECOVERY] step=%s prior_status=%s prior_run=%s prior_updated=%s reason=%s → force-like", self.step_id, decision.prior_status, decision.prior_run_id, decision.prior_updated_at, decision.reason, ) 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()) # ★유료 호출의 공통 길목에 **소유권 검사**을 건다. # # 모든 이미지 생성은 `reserve_current_call` 을 지난다. 확인을 각 호출부에 # 흩어 놓으면 반드시 빠지는 곳이 생기므로(한 씬 안에서만도 roll·재시도 # ·cine 변환이 각각 돈을 쓴다), 그 한 자리에서 본다. # # 거는 것은 취소 표 하나가 아니라 **`checkpoint_gate` 전체**다. 취소만 # 보면 `release` 로 락이 풀린 것과 다른 worker 가 가져간 것을 못 본다 # — 그 둘이야말로 좀비가 돈을 쓰는 갈래다. # # `strict=True` — 돈을 쓰기 직전이라 검사을 못 읽으면 **멈춘다.** # # 거는 자리를 **스텝 경계**로 잡은 이유: 스레드 지역이라 새 나가면 안 # 되는데, 스텝 하나가 시작하고 끝나는 이 자리가 정확히 그 수명이다. # 덤으로 이미지 스텝만이 아니라 **모든 스텝**이 덮인다. from app.core.image_call_budget import run_with_stop_check def _paid_call_gate() -> None: self.checkpoint_gate(where="유료 이미지 호출 직전", strict=True) with _step_trace_scope(self): return run_with_stop_check( _paid_call_gate, self._execute_and_finalize_inner, execute_mode, ) def _execute_and_finalize_inner(self, execute_mode: str) -> Dict[str, Any]: """`_execute_and_finalize` 의 본문 — 옮기기만 했다(내용 변경 0). 스텝 경계에 Opik trace 를 겹치려면 본문을 with 로 감싸야 하는데, 그러면 100줄 넘는 블록이 통째로 한 칸 들어가 diff 가 안 읽힌다. 그래서 본문을 메서드로 떼어 내고 바깥에서 감쌌다. ★`set_opik_context` local import 는 함께 옮긴다 — finally 절이 그것을 쓰는데, 바깥 함수의 지역 이름은 여기서 안 보인다(호출 시점까지 안 드러나는 결함이다). """ from app.modules.llm.llm_client import set_opik_context # 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 # 정지 요청으로 멈춘 것은 **실패가 아니다.** `failed` 로 적으면 # 운영자가 세운 것과 진짜로 깨진 것을 구별할 수 없다. `cancelled` 는 # STEP_STATUSES 에 이미 있는 값이다 (`step_manifest.py`). # # ★이 분기를 **별도 `except AppError` 절로 빼면 안 된다.** 그러면 # `_execute` 가 올린 다른 AppError 까지 그 절이 가로채고, 거기서 # re-raise 하면 아래 `failed` 기록이 통째로 건너뛰어진다 — # 형제 except 절은 이어서 안 돈다. 그래서 같은 절 안에서 가른다. # # ★`owner_lost` 는 여기 안 걸린다 (code 가 다르다). 이미 남의 것이 # 된 행에 쓰면 안 되므로 아래 owner-aware 경로가 처리한다. if ( isinstance(exc, AppError) and getattr(exc, "code", "") == "step.cancelled" ): logger.warning("[CANCELLED] step=%s — %s", self.step_id, exc.message) try: self._update_step_run_strict( "cancelled", error_message=exc.message ) except AppError as owner_exc: if owner_exc.code != "step.owner_lost": raise logger.warning( "Step %s cancel 기록 중 owner_lost: %s", self.step_id, owner_exc.message, ) raise # 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: # ★출발 상태를 넓힌다 (2026-08-26 Codex PR#4 BLOCK-2). # 마무리가 completed/partial 을 **이미 적은 뒤** 체크포인트 # 저장이 실패해 여기로 올 수 있다. 그때 'running' 만 # 받으면 되돌리기가 막혀 **DB 는 성공인데 실제로는 실패**인 # 채 남고, 다음 재개가 이 스텝을 건너뛴다. # 밖에서 'stale' 로 갈아 끼운 줄은 여전히 못 덮는다. self._update_step_run_strict( "failed", error_message=str(exc), expected_status=self._TERMINAL_ROLLBACK_FROM, ) 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' 격상 직전. ★한 줄도 안 바뀌면 **내 토큰이 이미 버려진 것**이다 (2026-08-26 Codex 재리뷰 BLOCK-4). 종전에는 `run_id` 조건만 걸고 결과를 안 봐서, 토큰이 버려진 뒤에도 조용히 0줄로 끝나고 바로 아래 `_get_recovery_count` 가 **새 주인의 줄**을 읽었다. 남의 카운터로 내 한계를 판단한 셈이다. """ now = self._now() res = 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 AND run_id = :rid AND status = 'running' """), {"reason": reason[:2000], "pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "now": now, "rid": self.run_id}) self.db.commit() # ★`run_id` 만 보면 **토큰을 안 돌린 외부 status 전이**를 못 잡는다 # (2026-08-26 Codex 2차 재리뷰). status 까지 걸어야 「내 것이고 # 아직 도는 중」이 확인된다. self._require_owned_row(res, "recovery_count 증가") return self._get_recovery_count() def _require_owned_row(self, result: Any, what: str) -> None: """방금 UPDATE 가 **내 줄**을 건드렸는지 확인한다. 0줄이면 내 토큰이 버림됐다는 뜻 — 조용히 넘어가면 그 뒤 판단이 전부 남의 줄을 근거로 이뤄진다. 여기서 멈춘다. """ rowcount = getattr(result, "rowcount", None) if rowcount == 0: raise AppError( code="step.owner_lost", message=( f"{self.step_id} {what} 실패 (run_id={self.run_id}) — " "이 주행의 토큰이 버림됐다. 다른 주행이 이 스텝을 가져갔거나 " "정지·풀기로 놓았다." ), status_code=409, ) def _reset_recovery_counter(self) -> None: """status='completed' 정상 진행 시 카운터 reset (다음 sweep 깨끗한 상태). ★부르는 자리가 `_update_step_run_strict(final_status)` **뒤**이고 `final_status == "completed"` 일 때뿐이다. 그러니 이 시점의 DB status 는 'completed' 여야 한다 — 아니면 그 사이에 누가 끼어든 것이다 (2026-08-26 Codex 2차 재리뷰). """ now = self._now() res = 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 AND run_id = :rid AND status = 'completed' """), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "now": now, "rid": self.run_id}) self.db.commit() self._require_owned_row(res, "recovery_count 초기화") def _get_recovery_count(self) -> int: # ★`run_id` 로 좁힌다 — 안 좁히면 토큰이 버려진 뒤 **새 주인의 카운터** # 를 읽어 내 한계를 판단한다. row = self.db.execute(text( "SELECT recovery_count FROM step_run " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid " " AND run_id = :rid" ), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "rid": self.run_id}).fetchone() return row[0] if row else 0 def _get_last_recovery_reason(self) -> str: # `_get_recovery_count` 와 같은 이유로 `run_id` 로 좁힌다. row = self.db.execute(text( "SELECT last_recovery_reason FROM step_run " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid " " AND run_id = :rid" ), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "rid": self.run_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 # ── v2 (2026-08-23): 키 이름을 litellm 에 맞추고 축을 가른다 ── try: from app.core.config import settings if getattr(settings, "opik_trace_v2_enabled", False): from app.modules.llm.opik_trace import ( build_axis_tags, episode_thread_id) # 죽은 키를 뺀다. litellm 은 trace_name 을 안 읽고 # session_id 도 안 본다(thread_id 를 본다). meta.pop("trace_name", None) meta.pop("session_id", None) meta.pop("project_name_tag", None) meta.pop("episode_title_tag", None) meta["thread_id"] = episode_thread_id( project_name=ctx.get("project_name", ""), episode_title=ctx.get("episode_title", ""), episode_id=self.episode_id or "", ) # 이름은 태그가 아니라 metadata 로 — 한글이고 카디널리티가 크다. # ★키 이름이 `project_name` 이면 **안 된다**(2026-08-24 E2E 실측). # 그것은 litellm 예약 키다 — `opik.py:47-50` 이 그 값을 # **Opik 프로젝트 이름**으로 읽어, 우리 파이프라인 프로젝트 # 이름마다 Opik 프로젝트가 새로 생기고 호출이 통째로 그리로 # 빠진다. 스텝 trace 는 `settings.opik_project_name` 쪽에 # 남으므로 부모와 자식이 갈라져 **계층이 영영 안 선다** # (실측: 스텝 trace 9개가 전부 span 0 · 호출 19건은 새 # 프로젝트에 `chat.completion` 으로). if ctx.get("project_name"): meta["pipeline_project_name"] = ctx["project_name"] if ctx.get("episode_title"): meta["episode_title"] = ctx["episode_title"] if ctx.get("run_tag"): meta["run_tag"] = ctx["run_tag"] # 태그는 축만. extra_tags 는 status 축으로 받는다 # (호출자가 "retry"·"gpt_fallback" 같은 낮은 카디널리티만 준다). axis = build_axis_tags(step=self.step_id) for t in (extra_tags or []): tag = f"status:{t}" if tag not in axis: axis.append(tag) meta["tags"] = axis except Exception as exc: # noqa: BLE001 logger.debug("build_opik_metadata v2 실패 (non-fatal): %s", exc) return meta @staticmethod def _now() -> str: return datetime.now(timezone.utc).isoformat()