jmt dZddlZddlZddlZddlZddlmZejeZ ddl m Z m Z m Z mZddlmZmZmZddlmZddlmZddlmZmZmZmZdd lmZdd lm Z dd l!m"Z#m$Z%m&Z&m'Z'm(Z(m)Z)dd l*m+Z+ed dgZ,dede-de-de.fdZ/e,jadeeeeeefde-de-dede+fdZ1e,jadeeeeeefde-de-de-dede+f dZ2e,jgdeddedd eddd!"ed#d$%eeeeeefde-d&e-d'e-d(ee4d)e5de-dede+fd*Z6e,jod+eed,ed,eeeeeefde-d-e4d.e4de-dede+f d/Z8de-de-defd0Z9ejtd1Z;e,jad2eeeedd3%eeeefde-de-dee-de+def d4Z%eeeefde-de-d?e-dee-de+def d@Z>e,jgdAed,dBed,dBedCdDeeeeeefde-d-e4d.e4d&e-de-dede+fdEZ?e,jadFeeeeeeefde-de-dede+fdGZ@e,jgdHeeddI%eddJ%eddK%eeeeeefde-dLe-dee-dMee-de-dede+fdNZAe,jgdOeeeeeeefde-de-dede+fdPZBe,jgdQeed,dR%ed#dS%eddT%eeeeeefde-de-dMe-dCe5dLe-de-dede+fdUZCde-fdVZDe,jgdWeddeeeeeefde-de-d&e-de-dede+f dXZEd_dYdZde-de-d[e5ddfd\ZFde-de-ddfd]ZGd_d^ZHy)`u@파이프라인 Step API — 개별 단계 실행/상태 조회.NPath)AnyDictListOptional) APIRouterDependsQuery)text)Session) api_endpointget_dbget_current_userverify_project_access)atomic_write_json)AppError)containsget_all_downstream_recursiveget_consumers_ofget_manifest_dictget_ordered_entriesget_resume_sensitive_step_ids) UserAccountz9/api/v1/projects/{project_id}/episodes/{episode_id}/stepssteps)prefixtagsdb project_id episode_idreturnc"ddlm}||||S)uGOpik thread 그룹핑용 context 생성 — dispatch_service로 위임.r)build_opik_context)&app.services.analysis_dispatch_servicer#)rrr r#s F/Users/manta/Documents/Projects/TheRoad-I1/backend/app/api/v1/steps.py_build_opik_contextr& sI b*j 99 current_userc&ddlm}d||||iS)u전체 단계 상태 + DAG 의존성 조회. W1-F11: 계산 로직은 `step_readmodel_service.get_all_steps_view`로 이관. r)get_all_steps_viewr)#app.services.step_readmodel_servicer+)r rrr)r+s r% get_all_stepsr-&sG 'J C DDr'z/{step_id}/resultstep_idcddl}ddlm}ddlm}t |st dd|d||j|z d z d z |z |z d z }|js|d dd S |j|jd} || jdd| d S#t$r} t dt| dd} ~ wwxYw)u8개별 단계 결과 조회 (체크포인트 데이터).rNrsettingszstep.not_foundu알 수 없는 단계: codemessage status_code checkpointsepisodesz manifest.jsonno_data)r.statusresultzutf-8)encodingr:unknownzstep.read_errori)jsonpathlibrapp.core.configr1_step_containsr projects_direxistsloads read_textget Exceptionstr) r r.rrr)r>rr1cp_pathdataexcs r%get_step_resultrL5s( ' ",8OPWy6Ygjkk X " "#j0  $ %'1 24; <>M N  >> "i4HHRtzz'++W+=>"dhhx.KW[\\ R-s3xSQQRs(7B C)CCz/run-allresumez^(resume|force)$)patternanalysisz^(analysis|image|all)$zW20E5: 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.)ge descriptionFzkW20E5: explicit operator approval to perform real image generation. Required for the shot-aware image path.)rQmodecategoryimage_call_capapprove_image_generationc ,ddlm}||||||||S)u전체 단계 순차 실행. 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 캡으로 동작). r)dispatch_category_run)rTrU)r$rW) r rRrSrTrUrrr)rWs r% run_all_stepsrXSs+FM   %!9 r'z/shot_selection/toggle. scene_index shot_indexcBddlm}||||j||S)uGShot 선택/해제 토글 — Phase 2.2 ShotSelectionService로 위임.r)ShotSelectionService)#app.services.shot_selection_servicer\toggle)r rYrZrrr)r\s r%toggle_shot_selectionr_s%I J ; B B;PZ [[r'cPddlm}t|j|z dz dz |z S)Nrr0r7r8)r@r1rrB)rr r1s r%_cp_baseras+( %% & 3m Cj PS] ]]r'z2^manifest_(\d{8}_\d{6})(?:_[a-zA-Z0-9_-]+)?\.json$z /snapshotsu특정 단계만 조회cBddlm}||||j|S)uF체크포인트 스냅샷 목록 조회 — SnapshotService로 위임.rSnapshotService)r.)app.services.snapshot_servicerd list_versions)r rr.r)rrds r%list_snapshotsrgs$> 2z: 6 D DW D UUr'u-특정 단계만 스냅샷. 없으면 전체.z^[a-zA-Z0-9_-]{0,32}$u스냅샷 라벨)rNrQlabelcDddlm}||||j||S)uV현재 체크포인트를 수동 스냅샷으로 저장 — SnapshotService로 위임.rrc)r.rh)rerd save_snapshot)r rr.rhr)rrds r%create_snapshotrks'> 2z: 6 D DW\a D bbr'z/snapshots/restorez ^\d{8}_\d{6}$u"복원할 버전 (YYYYMMDD_HHMMSS)uB특정 단계만 복원. 없으면 해당 버전의 전체 단계.versioncDddlm}||||j||S)uV특정 버전의 스냅샷으로 체크포인트 복원 — SnapshotService로 위임.rrc)rlr.)rerdrestore)r rrlr.r)rrds r%restore_snapshotros'> 2z: 6 > >wX_ > ``r'z/scene-detail/redo-shot)rPforcez ^(force)$c*ddlm}|||||||S)u단건 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 그대로. r)redo_scene_detail_shot)rrr rYrZrR)&app.services.scene_detail_redo_servicerr)r rYrZrRrrr)rrs r%redo_scene_detail_shot_endpointrts'(N !    r'z/lockscddlm}ddlm}m}m}|j td||dj}|\} } } g} |D]M} |j| j| j| j| j| jd}||||| jf|j |j"\}}| j%| j| j&| j| j(| jr| jj+nd| j,r| j,j+nd| j| j| jd |j.|| j0| j2d Pdd lm}||||}| | | | d |j8|j;d d S)u지금 누가 무엇을 잡고 있고, **살아있는가**. 판정은 추측이 아니다 — 같은 프로세스면 등록부를, 같은 호스트면 PID 를 직접 물어본다 (`app/core/step_lock.py` 머리말의 판정 순서). 응답의 `verdict` 는 셋이다: - `alive` 일하는 중이다. 건드리면 안 된다. - `dead` 소유 프로세스가 없다. `release` 로 풀어도 된다. - `unknown` 확정 못 한다. 사람이 봐야 한다. rr0) LockOwner judge_ownerprocess_identityaS 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 )pideid owner_host owner_pid owner_boot_id heartbeat_atrun_idkey lease_secondsallow_heartbeat_stealN)hostryboot_id) r.r:r started_atrcancel_requested_atownerverdictverdict_reasonrecovery_countlast_recovery_reason)read_cancel_state) requesteddescribe)locks this_process run_cancel)r@r1app.core.step_lockrvrwrxexecuter fetchallfrom_rowr|r}r~rrr.step_lock_lease_seconds!step_lock_heartbeat_steal_enabledappendr:r isoformatrvaluerrapp.core.run_controlrrr)r rrr)r1rvrwrxrowsrryrrrowrrreasonr cancel_states r%list_step_locksrs$)KK ::d Z 0 23;(* *+D#w E"".. ..,,jj $  & Z5"::"*"L"L    {{jjjj..030@0@  **,d**''11304}},, }}$!00$'$<$<+  L7$RZ@L!%cgF%//$--/  r'z/cancelu(왜 세우는가 — 기록에 남는다uU특정 스텝만 세운다. 비우면 이 에피소드의 주행 전체를 세운다.ustep_id 를 줄 때는 **필수**. GET /locks 에서 본 run_id 와 다르면 거부한다 — 그 사이 그 스텝이 끝나고 다음 주행이 잡았을 수 있다.rexpected_run_idc zddlm}|rt|stdd|d|stddd |rd nd }|j t d |d |||t d|rd|ini} |jt| j} tjd|| |j|xsdd|| | rddSddS|||||t|j} |j t d||t d|jd| j| jdS)u?**협조적 정지** — 멈추라는 표만 찍는다. 일하는 쪽은 안전 지점(샷 경계, provider 호출 직전)에서 이 표를 읽고 깨끗이 멈춘다. ★진행 중인 provider 호출을 중간에 끊지 않는다. 이미 접수한 이미지는 결과까지 가져온 뒤 멈춘다 — 끊어 봐야 돈은 이미 나갔고 결과만 잃는다. ★`step_id` 를 비우면 **주행 전체**를 세운다. 스텝 하나만 세우면 배치가 다음 스텝으로 넘어가므로 주행을 멈추려면 이쪽을 써야 한다. r)request_cancelz step.unknownu모르는 스텝입니다: r2r3zstep.cancel_needs_run_idustep_id 를 지정할 때는 expected_run_id 도 함께 주세요. GET .../steps/locks 에서 지금 run_id 를 확인하세요.izAND run_id = :expected_run_idr(a 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' z )ryrzsidnowrz/[STEP-CANCEL] step=%s marked=%s by=%s reason=%s-stepu7정지 표를 찍었다. 안전 지점에서 멈춘다.uJ지금 running 인 행이 없다 — 이미 끝났거나 시작 전이다.)scoper.markedr5)r requested_byz UPDATE step_run SET cancel_requested_at = CURRENT_TIMESTAMP, updated_at = :now WHERE project_id = :pid AND episode_id = :eid AND status = 'running' )ryrzrrun)rrr5)rrrArrr _now_isocommitboolrowcountloggerwarningidrHrr) r rr.rrrr)r run_id_clauseupdatedrstates r% cancel_runrQsB4g&#5gY? /T  ># r'z /cancel/clearcddlm}m}||||}|jsdddS|||||j}||rddSddS) u정지 요청을 내린다 — 다시 주행할 수 있게. ★이걸 안 부르면 정지 표가 그대로 남아 다음 주행도 즉시 선다. `run` 으로 새 claim 을 하면 그 스텝의 표는 지워지지만, 주행 단위 표는 여기서만 내려간다. r) clear_cancelrFu)걸려 있는 정지 요청이 없었다.)clearedr5)expected_requested_atu8정지 요청을 내렸다. 다시 주행할 수 있다.uZ그 사이 새 정지 요청이 들어왔다 — 내리지 않았다. 다시 확인하라.)rrrr requested_at)r rrr)rrseenrs r%clear_run_cancelrsoE RZ 8D >> -XYY J "//G  G  n  r'z/locks/{step_id}/releaseuPGET /locks 에서 본 run_id. 그 사이 주인이 바뀌었으면 거부된다.u소유자가 살아있다고 판정돼도 해제한다. 「그 프로세스는 내가 이미 죽였다」를 운영자가 선언하는 경우에만.u%왜 푸는가 — 기록에 남는다c ddlm}ddlm} m} m} |j td|||dj} | td|dd | jd k7rd | jd | jddS| j|k7rtdd|d| jdd | j| j| j| j| j | jd} | | |||f|j"|j$\}}|| j&ur#|s!td|d|j(d|dd dt+j,}d|j.d|j(d|d|d |xs| }|j td!|dd"|t1||||d#}|j3t5|j6}t8j;d$|||||j(||rd%d'Sd&d'S)(uT죽은 락을 푼다 — `running` → `failed`. ★**행을 지우지 않는다.** 죽은 것을 죽었다고 적는 것이다. 다음 `resume` 이 바로 가져간다. ★`expected_run_id` 를 요구하는 이유 — 조회한 뒤 해제하기까지 사이에 주인이 바뀌었을 수 있다. 그 값을 확인하지 않으면 **새로 들어온 산 주인을 죽인다.** ★기본은 죽음이 **확인될 때만** 통과한다. 살아있거나 확정 못 하면 거부하고, 운영자가 `force=true` 로 책임을 명시할 때만 통과시킨다. rr0)rv OwnerVerdictrwz 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 )ryrzrNzstep.lock_not_foundu 의 step_run 행이 없다.r2r3runningFu잡혀 있지 않다 (status=u) — 할 일이 없다.)releasedr:r5zstep.lock_movedu.그 사이 주인이 바뀌었다 (본 run_id=u, 지금 run_id=u$). 다시 조회하고 판단하라.ir{rzstep.lock_owner_not_deadu: 소유자가 죽었다고 확인되지 않는다 (판정=z: uK). 정말 그 프로세스를 죽였다면 force=true 로 다시 부르라.z released-zmanual release by z (verdict=z, force=z, revoked_run_id=z): a 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 i)noterevoked_run_idrryrzrru)[LOCK-RELEASE] step=%s released=%s — %su*풀었다. 다음 resume 이 가져간다.uE그 사이 상태가 바뀌어 풀지 않았다. 다시 조회하라.)rrrr5)r@r1rrvrrwrr fetchonerr:rrr|r}r~rrrDEADruuiduuid4rrrrrrr)r r.rrprrrr)r1rvrrwrrrrrrrrs r%release_step_lockrsF)GG **T Z @  B CK(*  {&i;<   zzYjj6szzlBZ[   zz_$"@@QR!!$ ,PR      nn]]**((**   E* W -66&HH G^l'''+)"==/N+;<[[  (!/N \__-.MM?(5'2)*#  #^ $ & jj   Ud (z! G*IIKG$$%H NN>SWX==( 9  Y  r'cdddlm}m}|j|jj S)uG`step_run` 의 text 시각 칸에 넣는 값 (기존 계약 그대로).rdatetimetimezone)rrrutcrrs r%rrts!+ << % / / 11r'z /{step_id}c ddlm}t|||}|||||||j|}|j d||S)u개별 단계 실행. mode: resume (기본) | force (재실행). 백그라운드 스레드. W1-F11: 실제 orchestration은 `step_execution_service.start_step`로 이관. route는 Opik context만 빌드한 뒤 서비스에 위임. r) start_step)rrr r.rRactor_id opik_contextr.)#app.services.step_execution_servicerr&r setdefault) r r.rRrrr)rrr;s r%run_stepr}sO?&r:zBL   !F i) Mr'TrrcDddlm}||||j|y)uT2I appearance count projection — Phase 2.1 EpisodeProjectionService로 이관. Legacy wrapper: 기존 호출자(entities.py) 호환 유지용. 신규 코드는 `app.services.checkpoint_sync.EpisodeProjectionService.sync_t2i_appearance_counts` 직접 사용. r)EpisodeProjectionServicerN)app.services.checkpoint_syncrsync_t2i_appearance_counts)rr rrrrs r%_sync_t2i_appearance_countsrs# FRZ8SS[aSbr'c$ddlm}||||y)u분석 체크포인트 → DB 동기화 (Phase 2.1 Service 오케스트레이터로 위임). Legacy wrapper: 기존 호출자(image_steps.py, steps.py run_step 등) 호환 유지용. 실제 로직은 `app.services.checkpoint_sync.orchestrate_full_sync`에 5-way 분해됨. r)orchestrate_full_syncN)rr)rr rrs r%_sync_checkpoints_to_dbrs C*j"5r'c(ddlm}|||||||S)uStep ID → StepRunner 인스턴스. Legacy wrapper: `app.services.analysis_dispatch_service.get_step_runner`로 위임. 기존 호출자(test_sync_v3.py, test_pipeline_v3_e2e.py, 내부 run_step) 호환 유지. r)get_step_runner)r$r)r.rr rproject_configrrs r%_get_step_runnerrs G 7J BP\ ]]r')N)I__doc__r>loggingrerr?r getLogger__name__rtypingrrrrfastapir r r sqlalchemyr sqlalchemy.ormr OrmSession app.api.depsrrrrapp.core.checkpoint_iorapp.core.errorsrapp.core.step_catalogrrAr$catalog_get_all_downstream_recursiverrrrapp.models.catalogrrouterrHdictr&rFr-rLpostintrrXpatchr_racompile _SNAP_TS_RErgrkrortrrrrrrrrrr'r%rsmF     8 $,,--0VV4$+ U]d\e f:J:C:S:T: B34V_ '(8 9 E E E  E E E  34V_ '(8 9 RR RR  R  R!R:Zh(:;*.FG$)  - %&+  B&34V_ '(8 9/,, ,,SM ,#,*+,, -,./,,^&' SzCj34V_ '(8 9 \ \ \ \ \  \  \( \$^^#^$^ bjj9 L 34"45NO '(8 9V_ V V Vc] V V  V V\ 34"45de /GUgh '(8 9V_ c c cc] c C= c  c  c c !" 34&6Dhi"45yz '(8 9V_ a a a ac] a  a  a# a &'SQ'CA&g{334V_ '(8 9      (\H 34V_ '(8 9 OOO O OOdY (RS" c &+  i&34V_ '(8 9%hh hc]hc]h !h" #h$%hhV_ 34V_ '(8 9       F '( ! ^   F(OP34V_ '(8 9'LL LL L !L"#L$ %L&'L)L^2#2\h(:;34V_ '(8 9        >ceicCcSc^bcnrc666T6^r'