"""파이프라인 Step API — 개별 단계 실행/상태 조회.""" import json import logging import re import uuid from pathlib import Path logger = logging.getLogger(__name__) from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, Query from sqlalchemy import text from sqlalchemy.orm import Session as OrmSession from app.api.deps import api_endpoint, get_db, get_current_user, verify_project_access from app.core.checkpoint_io import atomic_write_json from app.core.errors import AppError from app.core.step_catalog import ( contains as _step_contains, get_all_downstream_recursive as catalog_get_all_downstream_recursive, get_consumers_of, get_manifest_dict, get_ordered_entries, get_resume_sensitive_step_ids, ) from app.models.catalog import UserAccount router = APIRouter(prefix="/api/v1/projects/{project_id}/episodes/{episode_id}/steps", tags=["steps"]) def _build_opik_context(db: OrmSession, project_id: str, episode_id: str) -> dict: """Opik thread 그룹핑용 context 생성 — dispatch_service로 위임.""" from app.services.analysis_dispatch_service import build_opik_context return build_opik_context(db, project_id, episode_id) @router.get("") def get_all_steps( episode_id: str, project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """전체 단계 상태 + DAG 의존성 조회. W1-F11: 계산 로직은 `step_readmodel_service.get_all_steps_view`로 이관. """ from app.services.step_readmodel_service import get_all_steps_view return {"steps": get_all_steps_view(db, project_id, episode_id)} @router.get("/{step_id}/result") def get_step_result( episode_id: str, step_id: str, project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """개별 단계 결과 조회 (체크포인트 데이터).""" import json from pathlib import Path from app.core.config import settings if not _step_contains(step_id): raise AppError(code="step.not_found", message=f"알 수 없는 단계: {step_id}", status_code=404) cp_path = ( Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / step_id / "manifest.json" ) if not cp_path.exists(): return {"step_id": step_id, "status": "no_data", "result": None} try: data = json.loads(cp_path.read_text(encoding="utf-8")) return {"step_id": step_id, "status": data.get("status", "unknown"), "result": data} except Exception as exc: raise AppError(code="step.read_error", message=str(exc), status_code=500) @router.post("/run-all") def run_all_steps( episode_id: str, mode: str = Query("resume", pattern="^(resume|force)$"), category: str = Query("analysis", pattern="^(analysis|image|all)$"), image_call_cap: Optional[int] = Query( None, ge=0, description=( "W20E5: hard cap on provider image-generation calls for this " "run. Required (>= 1) when category in {image, all} and " "background_render_reference_mode == 'shot_aware_plan'. " "Ignored for category=analysis." ), ), approve_image_generation: bool = Query( False, description=( "W20E5: explicit operator approval to perform real image " "generation. Required for the shot-aware image path." ), ), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """전체 단계 순차 실행. category: analysis | image | all. Phase 4.1: dispatch 로직이 `analysis_dispatch_service`로 이관되어 `/episodes/{id}/analyze` 경로와 구현을 공유함. W20E5: ``image_call_cap`` + ``approve_image_generation`` 두 파라미터는 shot-aware 이미지 단계의 fail-closed 게이트에 연결된다. legacy 렌더 모드에서는 둘 다 옵션 (cap을 명시하면 advisory 캡으로 동작). """ from app.services.analysis_dispatch_service import dispatch_category_run return dispatch_category_run( project_id, episode_id, category, mode, db, image_call_cap=image_call_cap, approve_image_generation=approve_image_generation, ) @router.patch("/shot_selection/toggle") @api_endpoint def toggle_shot_selection( episode_id: str, scene_index: int = Query(...), shot_index: int = Query(...), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """Shot 선택/해제 토글 — Phase 2.2 ShotSelectionService로 위임.""" from app.services.shot_selection_service import ShotSelectionService return ShotSelectionService(db, project_id, episode_id).toggle(scene_index, shot_index) # ───────────────────────────────────────────────────────── # 체크포인트 스냅샷 — 저장 / 목록 / 복원 # (POST 라우트는 반드시 POST /{step_id} 위에 등록) # ───────────────────────────────────────────────────────── def _cp_base(project_id: str, episode_id: str) -> Path: from app.core.config import settings return Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id # 타임스탬프 추출용 (prerestore는 호출측에서 문자열 체크로 제외) _SNAP_TS_RE = re.compile( r"^manifest_(\d{8}_\d{6})(?:_[a-zA-Z0-9_-]+)?\.json$" ) @router.get("/snapshots") @api_endpoint def list_snapshots( episode_id: str, project_id: str = Depends(verify_project_access), step_id: Optional[str] = Query(None, description="특정 단계만 조회"), current_user: UserAccount = Depends(get_current_user), db: OrmSession = Depends(get_db), ): """체크포인트 스냅샷 목록 조회 — SnapshotService로 위임.""" from app.services.snapshot_service import SnapshotService return SnapshotService(db, project_id, episode_id).list_versions(step_id=step_id) @router.post("/snapshots") @api_endpoint def create_snapshot( episode_id: str, project_id: str = Depends(verify_project_access), step_id: Optional[str] = Query(None, description="특정 단계만 스냅샷. 없으면 전체."), label: Optional[str] = Query(None, pattern=r"^[a-zA-Z0-9_-]{0,32}$", description="스냅샷 라벨"), current_user: UserAccount = Depends(get_current_user), db: OrmSession = Depends(get_db), ): """현재 체크포인트를 수동 스냅샷으로 저장 — SnapshotService로 위임.""" from app.services.snapshot_service import SnapshotService return SnapshotService(db, project_id, episode_id).save_snapshot(step_id=step_id, label=label) @router.post("/snapshots/restore") @api_endpoint def restore_snapshot( episode_id: str, project_id: str = Depends(verify_project_access), version: str = Query(..., pattern=r"^\d{8}_\d{6}$", description="복원할 버전 (YYYYMMDD_HHMMSS)"), step_id: Optional[str] = Query(None, description="특정 단계만 복원. 없으면 해당 버전의 전체 단계."), current_user: UserAccount = Depends(get_current_user), db: OrmSession = Depends(get_db), ): """특정 버전의 스냅샷으로 체크포인트 복원 — SnapshotService로 위임.""" from app.services.snapshot_service import SnapshotService return SnapshotService(db, project_id, episode_id).restore(version=version, step_id=step_id) @router.post("/scene-detail/redo-shot") def redo_scene_detail_shot_endpoint( episode_id: str, scene_index: int = Query(..., ge=0), shot_index: int = Query(..., ge=0), mode: str = Query("force", pattern="^(force)$"), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """단건 shot scene_detail re-run — 운영 복구 도구. 2026-05-11 사용자 권장안 #2 축소형: scene_detail step 의 atomic-fail 대비. 1 shot 의 LLM 응답이 contract violation 으로 step 전체 abort 된 경우, 본 endpoint 로 그 shot 만 재실행. cp manifest 의 해당 shot 만 replace + checkpoint sync. UI 버튼 X — 운영 호출용. Returns: service.redo_scene_detail_shot 의 dict 그대로. """ from app.services.scene_detail_redo_service import redo_scene_detail_shot return redo_scene_detail_shot( db=db, project_id=project_id, episode_id=episode_id, scene_index=scene_index, shot_index=shot_index, mode=mode, ) # ══════════════════════════════════════════════════════════════════════════ # 멈추는 길 — 프로세스를 죽이지 않고 주행을 세운다 # # 이 시스템에는 주행을 멈출 방법이 프로세스를 죽이는 것밖에 없었다. 그런데 # 프로세스를 죽이면 `step_run` 에 `running` 행이 남고, 락은 소유자의 죽음을 # 확인할 길이 없어 경과 시간(기본 3600초)을 기다렸다 — 2026-08-26 새벽에 # 그렇게 40분을 잃었다. **멈추는 길이 없어서 락이 망가지는 순환**이었다. # # 세 갈래로 끊는다: # - `POST .../cancel` 협조적 정지 (돈 쓰는 중간에 안 끊는다) # - `POST .../locks/release` 죽은 락 풀기 (CAS — 본 run_id 를 요구한다) # - `GET .../locks` 진단 (누가 잡고 있고 살아있는가) # ══════════════════════════════════════════════════════════════════════════ @router.get("/locks") @api_endpoint def list_step_locks( episode_id: str, project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """지금 누가 무엇을 잡고 있고, **살아있는가**. 판정은 추측이 아니다 — 같은 프로세스면 등록부를, 같은 호스트면 PID 를 직접 물어본다 (`app/core/step_lock.py` 머리말의 판정 순서). 응답의 `verdict` 는 셋이다: - `alive` 일하는 중이다. 건드리면 안 된다. - `dead` 소유 프로세스가 없다. `release` 로 풀어도 된다. - `unknown` 확정 못 한다. 사람이 봐야 한다. """ from app.core.config import settings from app.core.step_lock import LockOwner, judge_owner, process_identity rows = db.execute(text(""" SELECT step_id, status, run_id, started_at, heartbeat_at, cancel_requested_at, owner_host, owner_pid, owner_boot_id, recovery_count, last_recovery_reason FROM step_run WHERE project_id = :pid AND episode_id = :eid AND status = 'running' ORDER BY started_at """), {"pid": project_id, "eid": episode_id}).fetchall() host, pid, boot_id = process_identity() locks = [] for row in rows: owner = LockOwner.from_row({ "owner_host": row.owner_host, "owner_pid": row.owner_pid, "owner_boot_id": row.owner_boot_id, "heartbeat_at": row.heartbeat_at, "run_id": row.run_id, }) verdict, reason = judge_owner( owner, key=(project_id, episode_id, row.step_id), lease_seconds=settings.step_lock_lease_seconds, allow_heartbeat_steal=settings.step_lock_heartbeat_steal_enabled, ) locks.append({ "step_id": row.step_id, "status": row.status, # release 를 부를 때 이 값을 그대로 돌려줘야 한다 (CAS). "run_id": row.run_id, "started_at": row.started_at, "heartbeat_at": ( row.heartbeat_at.isoformat() if row.heartbeat_at else None ), "cancel_requested_at": ( row.cancel_requested_at.isoformat() if row.cancel_requested_at else None ), "owner": { "host": row.owner_host, "pid": row.owner_pid, "boot_id": row.owner_boot_id, }, "verdict": verdict.value, "verdict_reason": reason, "recovery_count": row.recovery_count, "last_recovery_reason": row.last_recovery_reason, }) from app.core.run_control import read_cancel_state cancel_state = read_cancel_state(db, project_id, episode_id) return { "locks": locks, "this_process": {"host": host, "pid": pid, "boot_id": boot_id}, "run_cancel": { "requested": cancel_state.requested, "describe": cancel_state.describe(), }, } @router.post("/cancel") @api_endpoint def cancel_run( episode_id: str, reason: str = Query("", description="왜 세우는가 — 기록에 남는다"), step_id: Optional[str] = Query( None, description=( "특정 스텝만 세운다. 비우면 이 에피소드의 주행 전체를 세운다." ), ), expected_run_id: Optional[str] = Query( None, description=( "step_id 를 줄 때는 **필수**. GET /locks 에서 본 run_id 와 다르면 " "거부한다 — 그 사이 그 스텝이 끝나고 다음 주행이 잡았을 수 있다." ), ), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """**협조적 정지** — 멈추라는 표만 찍는다. 일하는 쪽은 안전 지점(샷 경계, provider 호출 직전)에서 이 표를 읽고 깨끗이 멈춘다. ★진행 중인 provider 호출을 중간에 끊지 않는다. 이미 접수한 이미지는 결과까지 가져온 뒤 멈춘다 — 끊어 봐야 돈은 이미 나갔고 결과만 잃는다. ★`step_id` 를 비우면 **주행 전체**를 세운다. 스텝 하나만 세우면 배치가 다음 스텝으로 넘어가므로 주행을 멈추려면 이쪽을 써야 한다. """ from app.core.run_control import request_cancel if step_id: if not _step_contains(step_id): raise AppError( code="step.unknown", message=f"모르는 스텝입니다: {step_id}", status_code=404, ) # ★스텝 하나를 세울 때는 **어느 주행인지** 밝혀야 한다. 안 밝히면 # 화면에서 본 스텝이 그 사이 끝나고 다음 주행이 같은 스텝을 잡았을 때 # 늦게 도착한 요청이 **엉뚱한 주행을 세운다.** if not expected_run_id: raise AppError( code="step.cancel_needs_run_id", message=( "step_id 를 지정할 때는 expected_run_id 도 함께 주세요. " "GET .../steps/locks 에서 지금 run_id 를 확인하세요." ), status_code=400, ) # ★`expected_run_id` 를 주면 **그 주행일 때만** 세운다. 화면에서 본 # 스텝이 그 사이 끝나고 다음 주행이 같은 스텝을 잡았을 수 있는데, # 확인하지 않으면 늦게 도착한 요청이 **엉뚱한 주행을 세운다.** run_id_clause = "AND run_id = :expected_run_id" if expected_run_id else "" updated = db.execute(text(f""" UPDATE step_run SET cancel_requested_at = CURRENT_TIMESTAMP, updated_at = :now WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid AND status = 'running' {run_id_clause} """), { "pid": project_id, "eid": episode_id, "sid": step_id, "now": _now_iso(), **({"expected_run_id": expected_run_id} if expected_run_id else {}), }) db.commit() marked = bool(updated.rowcount) logger.warning( "[STEP-CANCEL] step=%s marked=%s by=%s reason=%s", step_id, marked, current_user.id, reason or "-", ) return { "scope": "step", "step_id": step_id, "marked": marked, "message": ( "정지 표를 찍었다. 안전 지점에서 멈춘다." if marked else "지금 running 인 행이 없다 — 이미 끝났거나 시작 전이다." ), } state = request_cancel( db, project_id, episode_id, reason=reason, requested_by=str(current_user.id), ) # 지금 도는 스텝에도 같이 표를 찍는다 — 배치는 스텝 사이에서 서지만, # 도는 스텝은 자기 안의 안전 지점에서 서야 한다. db.execute(text(""" UPDATE step_run SET cancel_requested_at = CURRENT_TIMESTAMP, updated_at = :now WHERE project_id = :pid AND episode_id = :eid AND status = 'running' """), {"pid": project_id, "eid": episode_id, "now": _now_iso()}) db.commit() return { "scope": "run", "requested": state.requested, "message": state.describe(), } @router.post("/cancel/clear") @api_endpoint def clear_run_cancel( episode_id: str, project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """정지 요청을 내린다 — 다시 주행할 수 있게. ★이걸 안 부르면 정지 표가 그대로 남아 다음 주행도 즉시 선다. `run` 으로 새 claim 을 하면 그 스텝의 표는 지워지지만, 주행 단위 표는 여기서만 내려간다. """ from app.core.run_control import clear_cancel, read_cancel_state # ★내가 본 요청만 내린다. 조회하고 내리기까지 사이에 누가 **새 정지**를 # 요청했으면 그것은 남겨야 한다 — 「멈춰라」가 조용히 사라지는 것은 # 「멈추지 못하는 것」과 같다. seen = read_cancel_state(db, project_id, episode_id) if not seen.requested: return {"cleared": False, "message": "걸려 있는 정지 요청이 없었다."} cleared = clear_cancel( db, project_id, episode_id, expected_requested_at=seen.requested_at, ) return { "cleared": cleared, "message": ( "정지 요청을 내렸다. 다시 주행할 수 있다." if cleared else "그 사이 새 정지 요청이 들어왔다 — 내리지 않았다. 다시 확인하라." ), } @router.post("/locks/{step_id}/release") @api_endpoint def release_step_lock( episode_id: str, step_id: str, expected_run_id: str = Query( ..., description=( "GET /locks 에서 본 run_id. 그 사이 주인이 바뀌었으면 거부된다." ), ), force: bool = Query( False, description=( "소유자가 살아있다고 판정돼도 해제한다. 「그 프로세스는 내가 이미 " "죽였다」를 운영자가 선언하는 경우에만." ), ), reason: str = Query("", description="왜 푸는가 — 기록에 남는다"), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """죽은 락을 푼다 — `running` → `failed`. ★**행을 지우지 않는다.** 죽은 것을 죽었다고 적는 것이다. 다음 `resume` 이 바로 가져간다. ★`expected_run_id` 를 요구하는 이유 — 조회한 뒤 해제하기까지 사이에 주인이 바뀌었을 수 있다. 그 값을 확인하지 않으면 **새로 들어온 산 주인을 죽인다.** ★기본은 죽음이 **확인될 때만** 통과한다. 살아있거나 확정 못 하면 거부하고, 운영자가 `force=true` 로 책임을 명시할 때만 통과시킨다. """ from app.core.config import settings from app.core.step_lock import LockOwner, OwnerVerdict, judge_owner row = db.execute(text(""" SELECT status, run_id, started_at, heartbeat_at, owner_host, owner_pid, owner_boot_id FROM step_run WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid """), {"pid": project_id, "eid": episode_id, "sid": step_id}).fetchone() if row is None: raise AppError( code="step.lock_not_found", message=f"{step_id} 의 step_run 행이 없다.", status_code=404, ) if row.status != "running": return { "released": False, "status": row.status, "message": f"잡혀 있지 않다 (status={row.status}) — 할 일이 없다.", } if row.run_id != expected_run_id: raise AppError( code="step.lock_moved", message=( f"그 사이 주인이 바뀌었다 (본 run_id={expected_run_id}, " f"지금 run_id={row.run_id}). 다시 조회하고 판단하라." ), status_code=409, ) owner = LockOwner.from_row({ "owner_host": row.owner_host, "owner_pid": row.owner_pid, "owner_boot_id": row.owner_boot_id, "heartbeat_at": row.heartbeat_at, "run_id": row.run_id, }) verdict, verdict_reason = judge_owner( owner, key=(project_id, episode_id, step_id), lease_seconds=settings.step_lock_lease_seconds, allow_heartbeat_steal=settings.step_lock_heartbeat_steal_enabled, ) if verdict is not OwnerVerdict.DEAD and not force: raise AppError( code="step.lock_owner_not_dead", message=( f"{step_id} 소유자가 죽었다고 확인되지 않는다 " f"(판정={verdict.value}: {verdict_reason}). " "정말 그 프로세스를 죽였다면 force=true 로 다시 부르라." ), status_code=409, ) # ★풀기할 때 **소유 토큰을 버린다** (`run_id` 를 새 값으로 바꾼다). # # 상태만 `failed` 로 바꾸고 `run_id` 를 그대로 두면, 소유권 검사가 # `WHERE step_run.run_id = :rid` 하나뿐이라 (`step_runner._update_step_run`) # 옛 worker 의 진행률·마무리 갱신이 **같은 run_id 로 통과해 상태를 되살린다.** # 풀기해 놓고 다시 `running` 이 되는 것이다. # # 토큰을 바꾸면 그 worker 는 다음 갱신에서 owner_lost 로 떨어지고, # `checkpoint_gate` 도 run_id 불일치로 즉시 잡는다. 버림한 옛 토큰은 # 사유 문구에 남겨 나중에 읽을 수 있게 한다. revoked_run_id = f"released-{uuid.uuid4()}" note = ( f"manual release by {current_user.id} " f"(verdict={verdict.value}, force={force}, " f"revoked_run_id={expected_run_id}): " f"{reason or verdict_reason}" ) updated = db.execute(text(""" UPDATE step_run SET status = 'failed', run_id = :revoked_run_id, error_message = :note, last_recovery_reason = :note, recovery_count = COALESCE(recovery_count, 0) + 1, completed_at = :now, updated_at = :now WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid AND status = 'running' AND run_id = :run_id """), { "note": note[:2000], "revoked_run_id": revoked_run_id, "now": _now_iso(), "pid": project_id, "eid": episode_id, "sid": step_id, "run_id": expected_run_id, }) db.commit() released = bool(updated.rowcount) logger.warning("[LOCK-RELEASE] step=%s released=%s — %s", step_id, released, note) return { "released": released, "verdict": verdict.value, "verdict_reason": verdict_reason, "message": ( "풀었다. 다음 resume 이 가져간다." if released else "그 사이 상태가 바뀌어 풀지 않았다. 다시 조회하라." ), } def _now_iso() -> str: """`step_run` 의 text 시각 칸에 넣는 값 (기존 계약 그대로).""" from datetime import datetime, timezone return datetime.now(timezone.utc).isoformat() # ★`POST /{step_id}` 는 **맨 마지막**에 둔다. FastAPI 는 먼저 선언된 라우트를 # 먼저 맞춰 보므로, 이것이 위에 있으면 `POST .../cancel` 이 `step_id="cancel"` # 로 잡혀 정지 엔드포인트가 통째로 가려진다. @router.post("/{step_id}") def run_step( episode_id: str, step_id: str, mode: str = Query("resume", pattern="^(resume|force)$"), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """개별 단계 실행. mode: resume (기본) | force (재실행). 백그라운드 스레드. W1-F11: 실제 orchestration은 `step_execution_service.start_step`로 이관. route는 Opik context만 빌드한 뒤 서비스에 위임. """ from app.services.step_execution_service import start_step opik_context = _build_opik_context(db, project_id, episode_id) result = start_step( db=db, project_id=project_id, episode_id=episode_id, step_id=step_id, mode=mode, actor_id=current_user.id, opik_context=opik_context, ) # 기존 응답 호환: step_id 키 포함 result.setdefault("step_id", step_id) return result def _sync_t2i_appearance_counts(project_id: str, episode_id: str, db, logger=None, *, commit: bool = True) -> None: """T2I appearance count projection — Phase 2.1 EpisodeProjectionService로 이관. Legacy wrapper: 기존 호출자(entities.py) 호환 유지용. 신규 코드는 `app.services.checkpoint_sync.EpisodeProjectionService.sync_t2i_appearance_counts` 직접 사용. """ from app.services.checkpoint_sync import EpisodeProjectionService EpisodeProjectionService(db, project_id, episode_id).sync_t2i_appearance_counts(commit=commit) def _sync_checkpoints_to_db(project_id: str, episode_id: str, db) -> None: """분석 체크포인트 → DB 동기화 (Phase 2.1 Service 오케스트레이터로 위임). Legacy wrapper: 기존 호출자(image_steps.py, steps.py run_step 등) 호환 유지용. 실제 로직은 `app.services.checkpoint_sync.orchestrate_full_sync`에 5-way 분해됨. """ from app.services.checkpoint_sync import orchestrate_full_sync orchestrate_full_sync(project_id, episode_id, db) def _get_step_runner(step_id, project_id, episode_id, db, project_config, opik_context=None): """Step ID → StepRunner 인스턴스. Legacy wrapper: `app.services.analysis_dispatch_service.get_step_runner`로 위임. 기존 호출자(test_sync_v3.py, test_pipeline_v3_e2e.py, 내부 run_step) 호환 유지. """ from app.services.analysis_dispatch_service import get_step_runner return get_step_runner(step_id, project_id, episode_id, db, project_config, opik_context)