"""Step Execution Service — 개별 step 실행 오케스트레이션. W1-F11: `api/v1/steps.py.run_step`의 nested `_run_in_background` 본문과 validation/gate helper를 이관. route는 이 함수만 호출. 책임: - project_config 로드 - 동시 실행 충돌 체크 (run-all 중이면 409) - gate 확인 + disabled 차단 - resume skip 판단 - background worker 등록 (session 재생성 + pre/post sync) 비책임: - HTTP 파싱 / auth — route 담당 """ from __future__ import annotations import json import logging from typing import Any, Dict from sqlalchemy.orm import Session as OrmSession from app.core.errors import AppError from app.core.step_manifest import contains as _step_contains, get_manifest_dict logger = logging.getLogger(__name__) def _load_project_config(db: OrmSession, project_id: str) -> Dict[str, Any]: from app.models.project import ProjectSettings ps = db.query(ProjectSettings).filter(ProjectSettings.project_id == project_id).first() if not ps or not ps.llm_config_json: return {} try: return json.loads(ps.llm_config_json) except Exception as exc: logger.warning( "llm_config_json parse failed for project %s: %s — 기본 {} 사용", project_id, exc, ) return {} def _ensure_not_running_as_category(project_id: str, episode_id: str) -> None: """해당 에피소드에서 run-all이 진행 중이면 step 단독 실행 차단.""" from app.core.task_registry import is_task_running from app.services.analysis_dispatch_service import RUN_ALL_CATEGORY_TOKENS for cat in RUN_ALL_CATEGORY_TOKENS: if is_task_running(f"run_all:{project_id}:{episode_id}:{cat}"): raise AppError( code="step.pipeline_running", message="전체 파이프라인이 실행 중입니다", status_code=409, ) def _background_worker(pid, eid, sid, mode, config, actor_id, opik_ctx) -> None: """background thread entry — 자체 session 생성 + pre/post sync.""" from app.core.database import SessionLocal from app.services.analysis_dispatch_service import get_step_runner from app.services.checkpoint_sync import orchestrate_full_sync db2 = SessionLocal() try: entry_meta = get_manifest_dict(sid) or {} requires_presync = entry_meta.get("requires_projection_sync_before_run", False) if requires_presync: # Codex W1 High 2: flagged step은 presync 실패 시 실행 중단. # 잘못된 DB 상태로 run하면 조용히 잘못된 결과가 생긴다. try: orchestrate_full_sync(pid, eid, db2) except Exception as sync_exc: logger.error( "Required pre-sync failed before %s (aborting step run): %s", sid, sync_exc, ) db2.rollback() return bg_runner = get_step_runner(sid, pid, eid, db2, config, opik_context=opik_ctx) result = bg_runner.run(mode=mode) # partial cascade contract (problems.md #11) — single-step force 도 # run_steps_batch 와 동일하게 strict 검사. partial + manifest # allow_partial_downstream=False 면 episode.analysis_error 에 명시 # (downstream 사용자에게 가시화). step_run 자체는 step_runner 가 이미 # 'partial' 로 마킹 — 보존. result_status = (result or {}).get("status", "done") if result_status == "partial" and entry_meta.get("allow_partial_downstream", True) is False: logger.error( "Single-step %s partial and allow_partial_downstream=False — flagging episode", sid, ) try: from app.models.project import Episode ep = db2.query(Episode).filter(Episode.id == eid).first() if ep: ep.analysis_error = ( f"Step {sid} partial (allow_partial_downstream=False)" ) db2.commit() except Exception as flag_exc: logger.error("Failed to flag episode after partial-strict: %s", flag_exc) db2.rollback() # 개별 step 완료 후 DB projection 동기화 (분석 완료 상태면 반영). # W4 P3-2: step_id 전달 → step_run.sync_status 기록 (synced/failed). # 실패 기록은 orchestrator 내부에서 별도 세션으로 commit되므로, # 여기서 db2.rollback()해도 보존됨. episode.analysis_error는 보조 표시용. try: orchestrate_full_sync(pid, eid, db2, step_id=sid) except Exception as sync_exc: logger.error( "Post-step sync failed for %s — DB projection may be stale: %s", sid, sync_exc, ) db2.rollback() try: from app.models.project import Episode ep = db2.query(Episode).filter(Episode.id == eid).first() if ep: ep.analysis_error = f"Post-step sync stale after {sid}: {sync_exc}" db2.commit() except Exception: db2.rollback() except Exception as exc: logger.error("Step %s failed: %s", sid, exc) finally: db2.close() def start_step( db: OrmSession, project_id: str, episode_id: str, step_id: str, mode: str, actor_id: Any, opik_context: Dict[str, Any], ) -> Dict[str, Any]: """개별 step 실행 요청 처리. Returns: - {"ok": True, "step_id": sid, "status": "skipped", "reason": ...} (resume + 이미 완료 시) - {"ok": True, "status": "started", "job_key": ...} (background 등록 성공) Raises AppError (step.not_found / step.pipeline_running / step.disabled / step.already_running) — route에서 HTTP 변환. """ from app.core.job_manager import submit_background_job from app.services.analysis_dispatch_service import get_step_runner if not _step_contains(step_id): raise AppError(code="step.not_found", message=f"알 수 없는 단계: {step_id}", status_code=404) project_config = _load_project_config(db, project_id) _ensure_not_running_as_category(project_id, episode_id) # disabled 차단 step_meta = get_manifest_dict(step_id) or {} if step_meta.get("applicability") == "disabled": raise AppError(code="step.disabled", message=f"비활성화된 단계: {step_id}", status_code=400) # Gate 확인 (동기, 빠른 실패) runner = get_step_runner(step_id, project_id, episode_id, db, project_config, opik_context=opik_context) runner.check_gate() if mode == "resume": existing = runner._get_step_run(step_id) if existing and existing["status"] == "completed": # W6c: single-step resume preflight 를 StepRunner contract 에 위임. # 이전엔 compute_config_hash(project_config) 를 cp.config_hash 와 직접 # 비교해서 scene_detail 처럼 step-local _config_hash() 를 가진 step # 에서 false 409 를 만들었다 (project_config_hash 99914b932bd37a50 vs # step-local 187e36a52f3faa6a). StepRunner._evaluate_resume_decision # 은 step-local _config_hash 와 _check_cp_mismatch 통합 contract. from app.core.step_runner import ResumeAction decision = runner._evaluate_resume_decision("resume") if decision.action == ResumeAction.SKIP: return { "ok": True, "step_id": step_id, "status": "skipped", "reason": "already completed", } if decision.action == ResumeAction.BLOCK: raise AppError( code="step.resume_invalid", message=( f"체크포인트가 stale 입니다 ({decision.reason}). " f"force 모드로 재실행하세요." ), status_code=409, ) # 그 외 (RERUN_SELF artifact_missing/prior_state/contract_drift, # STALE_RUNNING_RECOVERY 등) → background submit 으로 fall through. # worker side 의 runner.run('resume') 이 동일 contract 로 분기를 # 재평가하므로 false skip 방지. logger.info( "start_step %s: completed cp decision=%s reason=%s — submitting background", step_id, decision.action.value, decision.reason, ) job_key = f"step:{project_id}:{episode_id}:{step_id}" started = submit_background_job( job_key=job_key, target=_background_worker, args=(project_id, episode_id, step_id, mode, project_config, actor_id, opik_context), description=f"Step {step_id} for {episode_id}", ) if not started: raise AppError(code="step.already_running", message=f"{step_id} 이미 실행 중", status_code=409) return {"ok": True, "status": "started", "job_key": job_key}