# Resume Architecture Fix — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Version**: v2.1.3 (Block A 누적 review NEEDS_REVISION — B1 _save_checkpoint_data fail-fast / B2 _review_entity_t2i fail-fast / I1 verify_completion fixture / I2 plan A5 result-level schema 정정) **Goal:** t2i_review → scene_detail cascade 폭주 사고 차단 + resume 구조 race 방지 + verify signal 분류로 자동 force 권한 제거. **Architecture:** 3 Block 직렬 (A: hotfix → **B0 dict shape 변경** → B: StepRunner decision 정리 → C: concurrency lock). dispatch 정지 유지. Block A+B+C 완료 후 dispatch 재개. Block D 는 별도 PR. **Tech Stack:** Python 3.12 / FastAPI / SQLAlchemy + raw SQL (PostgreSQL) / pytest / LiteLLM Router. spec: `docs/superpowers/specs/2026-05-07-resume-architecture-fix-design.md` **v5 (DRAFT_REVIEWED_PATCHED_V5)**. --- ## File Structure ### Block A — t2i_review hotfix - **Modify**: `backend/app/core/steps/_owned_helpers.py` — `refresh_t2i_prompt_hash()` primitive 추가 - **Modify**: `backend/app/core/steps/t2i_review_step.py` — `_refresh_scene_detail_sentinels()` / `_assert_card_hash_unchanged()` / `_execute()` 변경 - **Modify**: `backend/app/modules/pipeline/t2i_review.py` — `_build_item_id_map()` 추가, `_apply_scene_fixes()` 시그니처 변경, `run_t2i_review()` 반환 dict 확장 - **Modify**: `prompts/_base/t2i_review//scene_schema.json` — `item_id` required 추가 (새 prompt 버전 디렉토리, CLAUDE.md 규칙 — 덮어쓰기 금지). 디렉토리 위치는 repo root `prompts/` (backend/ 아님) - **Modify**: `prompts/_base/t2i_review//scene_system.md` — item_id protocol instruction 추가. 실제 load key = `scene_system` (`backend/app/modules/pipeline/t2i_review.py:216`), 확장자 `.md` - **Create**: `backend/tests/unit/test_owned_helpers_refresh.py` - **Create**: `backend/tests/unit/test_t2i_review_sentinel_refresh.py` - **Create**: `backend/tests/unit/test_t2i_review_item_id_map.py` ### Block B — StepRunner decision 정리 - **Modify**: `backend/app/core/integrity_report.py` — `CompletionReport.origin` 필드 추가 - **Modify**: `backend/app/core/step_runner.py` — V2 patch P4: `_get_step_run()` dict shape (Task **B0**, V5 S7). `ResumeAction` / `ResumeDecision` 추가, `_evaluate_*` 메서드 추가, V2 patch P3: `_execute_rerun_self()` / `_execute_force()` / `_execute_and_finalize()` (V5 S2 finalize 6 단계 보존), `_safe_verify_completion()` 변경, `_update_step_run(require_owner=True)` + `_update_step_run_strict()` (V2 추가 d / AC-C7), `run()` 흐름 재작성 (V5 S6 — AppError extra 금지), `_mark_not_applicable()` 추가 - **Modify**: `backend/app/services/step_execution_service.py` — 169-200 자체 skip 판단 제거 - **Modify**: `backend/app/core/steps/detail_steps.py` — V2 patch P5: `verify_completion()` 의 sentinel/card drift / loader / card recompute 분류 + origin 필드 배정 (V5 S4 — fixture 검증) - **Modify**: `backend/app/core/step_manifest.py` — `_LEGACY_SCHEMA_BUMP_ALLOWLIST` 추가 - **Modify**: `backend/app/core/config.py` — `step_running_timeout_seconds: int = 3600` 추가 - **Create**: `backend/tests/unit/test_get_step_run_shape.py` (V2 patch P4 — B0 신규) - **Create**: `backend/tests/unit/test_resume_decision.py` - **Create**: `backend/tests/unit/test_completion_report_origin.py` - **Create**: `backend/tests/unit/test_running_state_evaluation.py` - **Create**: `backend/tests/unit/test_legacy_schema_bump_allowlist.py` - **Create**: `backend/tests/unit/test_step_runner_owner_aware_update.py` (V2 추가 d — owner_lost test 포함) - **Create**: `backend/tests/unit/test_scene_detail_verify_origin.py` (V2 patch P5 — fixture 기반 7 branch) - **Create**: `backend/tests/integration/test_resume_single_vs_run_all_consistency.py` (V2 patch P6 — 두 path stub) ### Block C — Concurrency lock - **Modify**: `backend/app/core/step_runner.py` — `_try_claim_running()` 추가, `run()` 흐름에 claim 추가 - **Modify or Quarantine**: `backend/scripts/test_scene_detail.py` — V2 추가 c: 옵션 A (StepRunner.run 경유) or B (`backend/scripts/_quarantined/` 이동 + allowlist) 사용자 결정 - **Create**: `backend/tests/unit/test_try_claim_running.py` - **Create**: `backend/tests/integration/test_concurrent_claim.py` (V2 patch P6 — LLM call counter 검증 추가) - **Create**: `backend/scripts/audit_stale_running.py` — 배포 전 audit (V3 patch B1, app-level parse) --- ## Block A — t2i_review hotfix (~0.5d) ### ⚠️ Block A 진입 가드 (V2.1.2 — 실행 중 발견 lessons) **T3 ↔ T5 atomic dependency**: T3 (`_apply_scene_fixes` caller 전환 — `item_id` required) 와 T5 (`prompts/_base/t2i_review//scene_schema.json` + `scene_system.md` + `_review_scene_detail` producer 의 item_id propagation) 는 **반드시 같은 atomic boundary** 에서 commit 또는 같은 cycle 로 직렬 진행. 어느 한 쪽만 commit 하면 production t2i_review 가 첫 scene issue 에 `t2i_review.missing_item_id` 또는 `SchemaValidationError` raise 로 즉시 폭주. 실제 진행은 T3 commit → T5 stabilization commit 순서로 immediate sequence 권장 (별도 cycle 분리 금지). T5 가 **producer (items 수집 + user_prompt + result parsing) 전체 + integration test** 까지 포함해야 함 (plan A5 의 step 3 명시 외 producer 부분 추가 의무). **T5 schema 형식 deviation (result-level item_id)**: plan A5 의 schema 예시 (line 536-565) 는 `fixes.items` 단위 (issue-level item_id) 인데 producer 의 실제 result shape 은 `results[].issues[]` two-tier. **result-level 에 `item_id` required** 가 producer/schema align 으로 채택됨 (T5 v1 commit 602e44b). 향후 implementer 가 plan 의 `fixes.items` 형식 그대로 적용 시도 X — 실제는 result level. **T5 silent fallback 금지** (V2.1.2 stabilization v2 — commit 764d501): `result.item_id` 누락 / `SchemaValidationError` / batch failed 모두 fail-fast (`AppError(t2i_review.missing_item_id|batch_failed)` raise). `logger.warning + continue` 패턴은 `feedback_no_silent_fallback.md` 정책 위반. integration test 가 `test_t2i_review_run.py::test_run_t2i_review_raises_on_*` 3건 가드. ### Task A1: `_owned_helpers.refresh_t2i_prompt_hash()` primitive **Files:** - Modify: `backend/app/core/steps/_owned_helpers.py` (새 함수 추가) - Test: `backend/tests/unit/test_owned_helpers_refresh.py` (신규) **AC:** AC-A1 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_owned_helpers_refresh.py """refresh_t2i_prompt_hash primitive — AC-A1.""" import pytest from app.core.steps._owned_helpers import ( OWNED_SENTINEL_SCHEMA_VERSION, OWNED_VALIDATOR_FULL, compute_t2i_prompt_hash, compute_owned_hash, compute_camera_direction_hash, refresh_t2i_prompt_hash, ) from app.core.errors import AppError def _valid_sentinel(t2i: str = "old prompt") -> dict: return { "schema_version": OWNED_SENTINEL_SCHEMA_VERSION, "t2i_prompt_hash": compute_t2i_prompt_hash(t2i), "owned_hash": compute_owned_hash(["C01", "O02"]), "camera_direction_hash": compute_camera_direction_hash("dolly in"), "validator": OWNED_VALIDATOR_FULL, "violations": [], } def test_refresh_changes_only_t2i_prompt_hash(): sentinel = _valid_sentinel("old prompt") pre_owned = sentinel["owned_hash"] pre_camera = sentinel["camera_direction_hash"] pre_validator = sentinel["validator"] changed = refresh_t2i_prompt_hash(sentinel, "NEW prompt", where="test1") assert changed is True assert sentinel["t2i_prompt_hash"] == compute_t2i_prompt_hash("NEW prompt") assert sentinel["owned_hash"] == pre_owned assert sentinel["camera_direction_hash"] == pre_camera assert sentinel["validator"] == pre_validator def test_refresh_noop_when_already_matching(): sentinel = _valid_sentinel("same prompt") changed = refresh_t2i_prompt_hash(sentinel, "same prompt", where="test2") assert changed is False def test_refresh_raises_on_invalid_sentinel(): bad = {"schema_version": 1} # 다른 필드 없음 → shape 위반 with pytest.raises(AppError) as exc: refresh_t2i_prompt_hash(bad, "p", where="test3") assert "owned_validation" in str(exc.value).lower() or "missing" in str(exc.value).lower() ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_owned_helpers_refresh.py -v` Expected: FAIL — `cannot import name 'refresh_t2i_prompt_hash'` - [ ] **Step 3: primitive 구현** `backend/app/core/steps/_owned_helpers.py` 의 `compute_t2i_prompt_hash` 정의 직후 (line 121 부근) 추가: ```python def refresh_t2i_prompt_hash( sentinel: Dict[str, Any], t2i_prompt: str, where: str = "" ) -> bool: """sentinel 의 t2i_prompt_hash 만 갱신. 다른 hash 필드 보존. sentinel shape 검증 후 갱신 — shape 위반 시 raise. Returns: True (갱신 발생) / False (이미 일치, no-op). AC-A1 (spec §3.4): t2i_review 가 mutation 후 의무 호출. owned_hash / camera_direction_hash / validator / violations 는 보존. """ assert_owned_sentinel_shape(sentinel, where=where) new_hash = compute_t2i_prompt_hash(t2i_prompt) if sentinel["t2i_prompt_hash"] == new_hash: return False sentinel["t2i_prompt_hash"] = new_hash return True ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_owned_helpers_refresh.py -v` Expected: 3 passed - [ ] **Step 5: commit** ```bash git add backend/app/core/steps/_owned_helpers.py backend/tests/unit/test_owned_helpers_refresh.py git commit -m "feat(_owned_helpers): refresh_t2i_prompt_hash primitive (Block A AC-A1)" ``` --- ### Task A2: `t2i_review.py` pipeline — item_id map 빌더 **Files:** - Modify: `backend/app/modules/pipeline/t2i_review.py` (line 220 부근, 신규 helper) - Test: `backend/tests/unit/test_t2i_review_item_id_map.py` (신규) **AC:** AC-A6 (V3 patch I1) - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_t2i_review_item_id_map.py """item_id map builder — AC-A6 (V3 patch I1).""" from app.modules.pipeline.t2i_review import _build_item_id_map def test_build_item_id_map_basic(): scene_detail = { "scenes": [ {"scene_index": 1, "t2i_variations": [{"t2i_prompt": "a"}, {"t2i_prompt": "b"}]}, {"scene_index": 2, "t2i_variations": [{"t2i_prompt": "c"}]}, ] } m = _build_item_id_map(scene_detail) # format: f"S{scene_index}_L{scene_list_idx}_V{variation_idx}" assert m == { "S1_L0_V0": (0, 0), "S1_L0_V1": (0, 1), "S2_L1_V0": (1, 0), } def test_build_item_id_map_handles_fanout_same_scene_index(): """fan-out 시나리오 — 같은 scene_index 가 여러 entry.""" scene_detail = { "scenes": [ {"scene_index": 5, "t2i_variations": [{"t2i_prompt": "a"}]}, {"scene_index": 5, "t2i_variations": [{"t2i_prompt": "b"}]}, # 같은 scene_index, 다른 list 위치 ] } m = _build_item_id_map(scene_detail) # scene_list_idx 가 다르면 item_id 도 다름 (모호 X) assert m == { "S5_L0_V0": (0, 0), "S5_L1_V0": (1, 0), } def test_build_item_id_map_skips_scenes_without_variations(): scene_detail = { "scenes": [ {"scene_index": 1}, # t2i_variations 키 없음 {"scene_index": 2, "t2i_variations": []}, {"scene_index": 3, "t2i_variations": [{"t2i_prompt": "x"}]}, ] } m = _build_item_id_map(scene_detail) assert m == {"S3_L2_V0": (2, 0)} ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_item_id_map.py -v` Expected: FAIL — `cannot import name '_build_item_id_map'` - [ ] **Step 3: helper 구현** `backend/app/modules/pipeline/t2i_review.py` 상단 import 직후 (line 20 부근) 추가: ```python from typing import Tuple def _build_item_id_map(scene_detail_data: Dict) -> Dict[str, Tuple[int, int]]: """t2i_variations 모두 순회 — item_id (deterministic) → (scene_list_idx, variation_idx) map 생성. item_id format: f"S{scene_index}_L{scene_list_idx}_V{variation_idx}" LLM 이 prompt 의 item_id 를 echo 하면 application 이 map 으로 정확 위치 찾음. AC-A6 (V3 patch I1, spec §3.3): fan-out 시나리오 (같은 scene_index 다수) 모호성 차단. """ m: Dict[str, Tuple[int, int]] = {} for s_list_idx, scene in enumerate(scene_detail_data.get("scenes", [])): scene_index = scene.get("scene_index") for v_idx, _variation in enumerate(scene.get("t2i_variations", [])): item_id = f"S{scene_index}_L{s_list_idx}_V{v_idx}" m[item_id] = (s_list_idx, v_idx) return m ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_item_id_map.py -v` Expected: 3 passed - [ ] **Step 5: commit** ```bash git add backend/app/modules/pipeline/t2i_review.py backend/tests/unit/test_t2i_review_item_id_map.py git commit -m "feat(t2i_review): _build_item_id_map for deterministic mutation path (AC-A6)" ``` --- ### Task A3: `t2i_review.py` `_apply_scene_fixes` 시그니처 변경 + applied_indices 반환 **Files:** - Modify: `backend/app/modules/pipeline/t2i_review.py` (`_apply_scene_fixes` 시그니처 + 본문) - Test: `backend/tests/unit/test_t2i_review_apply_scene_fixes.py` (신규) **AC:** AC-A6 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_t2i_review_apply_scene_fixes.py """_apply_scene_fixes — item_id 기반 정확 path + applied_indices 반환 (AC-A6).""" import pytest from app.modules.pipeline.t2i_review import ( _apply_scene_fixes, _build_item_id_map, ) from app.core.errors import AppError def _scene_data(): return { "scenes": [ {"scene_index": 1, "t2i_variations": [{"t2i_prompt": "old1"}, {"t2i_prompt": "old2"}]}, {"scene_index": 2, "t2i_variations": [{"t2i_prompt": "old3"}]}, ] } def test_apply_returns_applied_indices_count(): scene_data = _scene_data() item_id_map = _build_item_id_map(scene_data) fixes = [ {"item_id": "S1_L0_V0", "target": "old1", "suggestion": "new1"}, {"item_id": "S2_L1_V0", "target": "old3", "suggestion": "new3"}, ] count, applied = _apply_scene_fixes(scene_data, fixes, item_id_map) assert count == 2 assert applied == [(0, 0), (1, 0)] # 실제 적용 검증 assert scene_data["scenes"][0]["t2i_variations"][0]["t2i_prompt"] == "new1" assert scene_data["scenes"][1]["t2i_variations"][0]["t2i_prompt"] == "new3" def test_apply_raises_on_unknown_item_id(): scene_data = _scene_data() item_id_map = _build_item_id_map(scene_data) fixes = [{"item_id": "S99_L99_V99", "target": "x", "suggestion": "y"}] with pytest.raises(AppError) as exc: _apply_scene_fixes(scene_data, fixes, item_id_map) assert exc.value.code == "t2i_review.unknown_item_id" def test_apply_skips_no_op_target_not_found(): """target 가 t2i_prompt 안에 없으면 적용 안 함 (count 미증가).""" scene_data = _scene_data() item_id_map = _build_item_id_map(scene_data) fixes = [ {"item_id": "S1_L0_V0", "target": "non_existent_token", "suggestion": "x"}, ] count, applied = _apply_scene_fixes(scene_data, fixes, item_id_map) assert count == 0 assert applied == [] ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_apply_scene_fixes.py -v` Expected: FAIL — 시그니처 mismatch 또는 import 에러 - [ ] **Step 3: `_apply_scene_fixes` 시그니처 변경** `backend/app/modules/pipeline/t2i_review.py` 의 `_apply_scene_fixes` 함수 (line ~290 부근) 를 다음으로 교체: ```python def _apply_scene_fixes( scene_detail_data: Dict, scene_fixes: List[Dict], item_id_map: Dict[str, Tuple[int, int]], ) -> Tuple[int, List[Tuple[int, int]]]: """LLM 이 schema required 로 echo 한 fix["item_id"] 로 정확 위치 매핑. Returns: (count, applied_indices) — count 는 실제 적용된 fix 수, applied_indices 는 (scene_list_idx, variation_idx) 튜플 리스트. AC-A6 (V3 patch I1, spec §3.3): fan-out 모호성 차단. Raises: AppError(code="t2i_review.unknown_item_id"): LLM 이 prompt 외 item_id 반환 시. """ from app.core.errors import AppError applied_indices: List[Tuple[int, int]] = [] count = 0 for fix in scene_fixes: item_id = fix.get("item_id") if not item_id: raise AppError( code="t2i_review.missing_item_id", message=f"fix 에 item_id 누락: {fix!r}", ) if item_id not in item_id_map: raise AppError( code="t2i_review.unknown_item_id", message=f"LLM 이 반환한 item_id={item_id!r} 가 prompt 에 없음 — schema 위반", ) s_list_idx, v_idx = item_id_map[item_id] scene = scene_detail_data["scenes"][s_list_idx] variation = scene["t2i_variations"][v_idx] target = fix.get("target", "") suggestion = fix.get("suggestion", "") old_prompt = variation.get("t2i_prompt", "") if target and target in old_prompt: variation["t2i_prompt"] = old_prompt.replace(target, suggestion) applied_indices.append((s_list_idx, v_idx)) count += 1 return count, applied_indices ``` 기존 caller (`run_t2i_review` 안의 호출 site, line ~61 부근) 도 함께 수정: ```python # 기존: s_count = _apply_scene_fixes(scene_detail_data, scene_fixes) # 새로: item_id_map = _build_item_id_map(scene_detail_data) s_count, scene_applied_indices = _apply_scene_fixes( scene_detail_data, scene_fixes, item_id_map, ) ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_apply_scene_fixes.py -v` Expected: 3 passed - [ ] **Step 5: commit** ```bash git add backend/app/modules/pipeline/t2i_review.py backend/tests/unit/test_t2i_review_apply_scene_fixes.py git commit -m "refactor(t2i_review): _apply_scene_fixes uses item_id, returns applied_indices (AC-A6)" ``` --- ### Task A4: `run_t2i_review()` 반환 dict 에 `scene_applied_indices` 추가 **Files:** - Modify: `backend/app/modules/pipeline/t2i_review.py` (`run_t2i_review` 의 return 부분, line ~70 부근) **AC:** AC-A6 - [ ] **Step 1: 기존 caller 동작 확인 + 테스트 추가** Append to `backend/tests/unit/test_t2i_review_apply_scene_fixes.py`: ```python def test_run_t2i_review_returns_scene_applied_indices(monkeypatch): """run_t2i_review 반환 dict 에 scene_applied_indices 포함 (AC-A6). V2 patch (추가 b — Codex IMPORTANT #1): _review_entity_t2i / _review_scene_detail 는 실제로 list 반환 (backend/app/modules/pipeline/t2i_review.py:50,55). stub 도 list 로 — 함수 계약 강제 변경 금지. violation summary 는 별도 spec 후 추가 (현 spec scope 외). """ from app.modules.pipeline import t2i_review as mod # _review_entity_t2i / _review_scene_detail 를 stub — 실제 계약대로 list 반환 monkeypatch.setattr(mod, "_review_entity_t2i", lambda *a, **kw: []) monkeypatch.setattr( mod, "_review_scene_detail", lambda *a, **kw: [ {"item_id": "S1_L0_V0", "target": "old1", "suggestion": "new1"}, ], ) entity_data = {"entities": []} scene_data = { "scenes": [ {"scene_index": 1, "t2i_variations": [{"t2i_prompt": "old1"}]}, ] } result = mod.run_t2i_review( entity_t2i_data=entity_data, scene_detail_data=scene_data, entity_merge_data={}, entity_detail_data={}, shot_extract_data={}, shot_staging_data={}, vwr_data={}, ) assert "scene_applied_indices" in result assert result["scene_applied_indices"] == [(0, 0)] assert result["scene_applied"] == 1 ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_apply_scene_fixes.py::test_run_t2i_review_returns_scene_applied_indices -v` Expected: FAIL — `scene_applied_indices` 키 없음 - [ ] **Step 3: `run_t2i_review` 반환 dict 확장** `backend/app/modules/pipeline/t2i_review.py` 의 `run_t2i_review` 본문 (line 55-75) 에서 `_apply_scene_fixes` 호출 후 return dict 수정: ```python # V2 patch (추가 b — Codex IMPORTANT #1): 현 함수 계약 보존 — 기존 키 그대로 유지하고 # scene_applied_indices 만 추가. _review_*_t2i 의 반환은 list (현 line 50/55) — tuple 변경 X. # 기존 (backend/app/modules/pipeline/t2i_review.py:65-74): # return { # "entity_fixes": len(entity_fixes), # "scene_fixes": len(scene_fixes), # "entity_applied": e_count, # "scene_applied": s_count, # "details": {"entity": entity_fixes, "scene": scene_fixes}, # } # # 새로 — scene_applied_indices 추가만: return { "entity_fixes": len(entity_fixes), "scene_fixes": len(scene_fixes), "entity_applied": e_count, "scene_applied": s_count, "scene_applied_indices": scene_applied_indices, # NEW (AC-A6) "details": {"entity": entity_fixes, "scene": scene_fixes}, } ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_apply_scene_fixes.py -v` Expected: 4 passed - [ ] **Step 5: commit** ```bash git add backend/app/modules/pipeline/t2i_review.py backend/tests/unit/test_t2i_review_apply_scene_fixes.py git commit -m "feat(t2i_review): expose scene_applied_indices in run_t2i_review return (AC-A6)" ``` --- ### Task A5: prompt schema + template 에 `item_id` required 추가 **V2 patch (P1 — Codex BLOCKING #1): prompt 디렉토리 path 정정**: - 잘못 (v1): `backend/prompts/_base/t2i_review/.../system.txt` — 디렉토리 부재 - 정정 (v2): `prompts/_base/t2i_review//scene_schema.json` + `prompts/_base/t2i_review//scene_system.md` - 실제 active prompt loader: `load_prompt("t2i_review", "scene_system")` (`backend/app/modules/pipeline/t2i_review.py:216`) — `.md` 확장자 **Files:** - Modify: `prompts/_base/t2i_review//scene_schema.json` (새 버전 디렉토리, 기존 덮어쓰기 금지 — CLAUDE.md 규칙) - Modify: `prompts/_base/t2i_review//scene_system.md` (실제 load key = `scene_system`) **AC:** AC-A6 - [ ] **Step 1: 현재 prompt 디렉토리 식별** ```bash ls prompts/_base/t2i_review/ # Expected: 1.202604051200/ 2.202604301730/ (또는 더 최신) ``` 가장 최신 버전 디렉토리 확인 (예: `2.202604301730/`). 새 버전 디렉토리 생성 (CLAUDE.md 규칙 — 덮어쓰기 금지): ```bash # 예시 — 현재 시간 기준 새 버전 (2026-05-08 기준) NEW_VER="2.202605081200" LATEST=$(ls -1 prompts/_base/t2i_review/ | grep -E "^[0-9]" | sort | tail -1) cp -r "prompts/_base/t2i_review/$LATEST" "prompts/_base/t2i_review/$NEW_VER" ls "prompts/_base/t2i_review/$NEW_VER/" # Expected (현재 v2.202604301730 기준): scene_schema.json, scene_system.md, entity_schema.json, entity_system.md ``` - [ ] **Step 2: scene_schema.json 수정 (result-level item_id required, 추가 e — scene_index/var_index optional 명시)** **V2.1.3 patch (Codex IMPORTANT #2 — Block A 누적 review)**: 옛 v2.1.1 본문은 `properties.fixes.items` (issue-level item_id) — producer (`_review_scene_detail`) 의 실제 result shape `results[].issues[]` two-tier 와 mismatch. v2.1.2 intro 가드 (line 57 부근) 가 result-level 채택을 명시했지만 본 step 2 의 schema 예시 가 잔재 — 본 patch 로 정정. 향후 implementer 가 옛 `fixes.items` 형식 그대로 적용 X. **채택**: `properties.results.items` 에 `item_id` required (result-level). `scene_index` / `var_index` 는 보조 정보 — properties 에는 두되 required 에서 제외 (V2 patch 추가 e — MINOR #2): ```json { "type": "object", "properties": { "results": { "type": "array", "items": { "type": "object", "properties": { "item_id": { "type": "string", "description": "S{scene_index}_L{scene_list_idx}_V{variation_idx} — prompt 에 명시된 deterministic ID. LLM echo only. SINGLE SOURCE OF TRUTH." }, "scene_index": { "type": "integer", "description": "보조 정보 (optional) — item_id 가 권위. LLM 이 누락해도 OK." }, "var_index": { "type": "integer", "description": "보조 정보 (optional) — item_id 가 권위. LLM 이 누락해도 OK." }, "has_issues": {"type": "boolean"}, "issues": { "type": "array", "items": { "type": "object", "properties": { "type": {"type": "string"}, "target": {"type": "string"}, "suggestion": {"type": "string"} }, "required": ["type", "target", "suggestion"] } } }, "required": ["item_id", "has_issues", "issues"] } } }, "required": ["results"] } ``` - [ ] **Step 3: scene_system.md 에 item_id protocol instruction 추가** `prompts/_base/t2i_review//scene_system.md` (실제 load key = `scene_system`) 본문에 다음 instruction 추가: ``` # ITEM_ID PROTOCOL (V3 patch I1) 각 t2i_variation 은 deterministic `item_id` (예: `S1_L0_V0`) 가 prefix 로 명시된다. fix 를 제안할 때 반드시 해당 item 의 `item_id` 를 그대로 echo 해야 한다. - scene_index/var_index 는 보조 정보로만 — item_id 가 single source of truth. - 같은 scene_index 가 여러 entry 일 수 있음 (fan-out) — item_id 로 구분. ``` scene_detail context injection 부분에서 각 variation 출력 시 `[item_id: S{si}_L{li}_V{vi}]` prefix 자동 첨부 (코드 변경 — `t2i_review.py` 의 prompt 구성 부분): ```python # t2i_review.py — _build_scene_detail_context 또는 prompt 구성 site 변경 # 각 variation 을 prompt 에 넣을 때 item_id prefix 추가 for s_list_idx, scene in enumerate(scenes): scene_index = scene.get("scene_index") for v_idx, variation in enumerate(scene.get("t2i_variations", [])): item_id = f"S{scene_index}_L{s_list_idx}_V{v_idx}" prompt_text += f"\n[item_id: {item_id}] {variation['t2i_prompt']}\n" ``` - [ ] **Step 4: 단위 테스트 — schema 검증 (V2.1.3 — result-level)** **V2.1.3 patch (Codex IMPORTANT #2)**: 옛 v2.1.1 verification script 는 `fix_props = data['properties']['fixes']['items']` 로 issue-level 검증 — producer shape 와 mismatch. result-level item_id 정합 검증으로 정정. ```bash # V2 patch P1: prompts/ 는 repo root 기준 — backend cwd 에서 ../prompts/ # V2.1.3 patch: result-level item_id (옛 fixes.items issue-level 폐기). cd backend && PYTHONPATH=. .venv/bin/python -c " import json from pathlib import Path schemas = Path('../prompts/_base/t2i_review').glob('*/scene_schema.json') for s in sorted(schemas)[-1:]: # 최신 버전만 data = json.loads(s.read_text()) result_props = data['properties']['results']['items'] assert 'item_id' in result_props['properties'], f'item_id 누락: {s}' assert 'item_id' in result_props['required'], f'item_id required 누락: {s}' # 추가 e (Codex MINOR #2): scene_index/var_index 는 properties 에는 있되 required 에서 제외 assert 'scene_index' not in result_props['required'], f'scene_index 는 optional 이어야: {s}' assert 'var_index' not in result_props['required'], f'var_index 는 optional 이어야: {s}' print(f'OK: {s}') " ``` Expected: `OK: ../prompts/_base/t2i_review//scene_schema.json` - [ ] **Step 5: commit** ```bash git add prompts/_base/t2i_review/ backend/app/modules/pipeline/t2i_review.py git commit -m "feat(t2i_review prompt): item_id required in schema + prompt prefix (AC-A6)" ``` --- ### Task A6: `t2i_review_step.py` — `_refresh_scene_detail_sentinels` 통합 **Files:** - Modify: `backend/app/core/steps/t2i_review_step.py` (line 55-62 mutation site) - Test: `backend/tests/unit/test_t2i_review_sentinel_refresh.py` (신규) **AC:** AC-A1, AC-A2, AC-A4, AC-A5 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_t2i_review_sentinel_refresh.py """t2i_review step 의 sentinel refresh 통합 — AC-A1, A2, A4, A5.""" from copy import deepcopy import pytest from app.core.steps._owned_helpers import ( OWNED_SENTINEL_SCHEMA_VERSION, OWNED_VALIDATOR_FULL, compute_t2i_prompt_hash, compute_owned_hash, compute_camera_direction_hash, ) from app.core.steps.t2i_review_step import _refresh_scene_detail_sentinels from app.core.errors import AppError def _sentinel(t2i: str) -> dict: return { "schema_version": OWNED_SENTINEL_SCHEMA_VERSION, "t2i_prompt_hash": compute_t2i_prompt_hash(t2i), "owned_hash": compute_owned_hash(["C01"]), "camera_direction_hash": compute_camera_direction_hash("static"), "validator": OWNED_VALIDATOR_FULL, "violations": [], } def test_refresh_updates_only_modified_indices(): """AC-A1: applied_indices 외 sentinel 은 보존.""" scene_data = {"scenes": [ {"t2i_variations": [ {"t2i_prompt": "NEW", "owned_validation": _sentinel("OLD")}, {"t2i_prompt": "untouched", "owned_validation": _sentinel("untouched")}, ]} ]} refreshed = _refresh_scene_detail_sentinels(scene_data, applied_indices=[(0, 0)]) assert refreshed == 1 # idx 0 갱신 assert scene_data["scenes"][0]["t2i_variations"][0]["owned_validation"]["t2i_prompt_hash"] == compute_t2i_prompt_hash("NEW") # idx 1 보존 (그대로) assert scene_data["scenes"][0]["t2i_variations"][1]["owned_validation"]["t2i_prompt_hash"] == compute_t2i_prompt_hash("untouched") def test_refresh_returns_zero_when_no_changes(): """AC-A5: t2i_prompt 가 이미 hash 와 일치하면 no-op.""" scene_data = {"scenes": [ {"t2i_variations": [ {"t2i_prompt": "same", "owned_validation": _sentinel("same")}, ]} ]} refreshed = _refresh_scene_detail_sentinels(scene_data, applied_indices=[(0, 0)]) assert refreshed == 0 def test_refresh_raises_when_owned_validation_missing(): """AC-A2: sentinel 누락 → t2i_review failed (raise).""" scene_data = {"scenes": [ {"t2i_variations": [ {"t2i_prompt": "x"}, # owned_validation 키 없음 ]} ]} with pytest.raises(AppError) as exc: _refresh_scene_detail_sentinels(scene_data, applied_indices=[(0, 0)]) assert exc.value.code in ( "t2i_review.owned_validation_missing", "t2i_review.sentinel_refresh_failed", ) ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_sentinel_refresh.py -v` Expected: FAIL — `_refresh_scene_detail_sentinels` 미정의 - [ ] **Step 3: helper + `_execute()` 수정** `backend/app/core/steps/t2i_review_step.py` 상단 import 추가: ```python from typing import Dict, List, Tuple, Optional from app.core.errors import AppError from app.core.steps._owned_helpers import refresh_t2i_prompt_hash ``` 파일 하단 (class 외부) 또는 class 안의 helper section 에 추가: ```python def _refresh_scene_detail_sentinels( scene_detail_data: Dict, applied_indices: List[Tuple[int, int]] ) -> int: """t2i_review 가 수정한 (scene_list_idx, variation_idx) 항목의 owned_validation sentinel 의 t2i_prompt_hash 만 갱신. AC-A1, A2 (spec §3.4): applied_indices 외 항목은 건드리지 않음. sentinel 누락 / shape 위반 시 raise → caller 가 t2i_review failed 처리. """ refreshed = 0 scenes = scene_detail_data["scenes"] for s_list_idx, v_idx in applied_indices: scene = scenes[s_list_idx] variation = scene["t2i_variations"][v_idx] sentinel = variation.get("owned_validation") if not sentinel: raise AppError( code="t2i_review.owned_validation_missing", message=( f"scenes[{s_list_idx}].t2i_variations[{v_idx}].owned_validation 없음 — schema 위반" ), ) if refresh_t2i_prompt_hash( sentinel, variation["t2i_prompt"], where=f"t2i_review_refresh@s{s_list_idx}_v{v_idx}", ): refreshed += 1 return refreshed def _capture_card_hash_state( scene_detail_data: Dict, ) -> Dict[Tuple[int, int], Optional[str]]: """mutation 직전 모든 (scene_idx, var_idx) 의 render_prompt_card_hash 전체 capture. V2 patch P2 (Codex BLOCKING #2 + spec §3.5 V5 patch S1): run_t2i_review() 호출 **전** capture 의무 — 호출 후엔 이미 mutation 끝남. applied_indices 는 mutation 후에야 알 수 있으므로 **전체** scenes/variations 매핑. under-coverage 위험 0 — applied_indices 는 항상 전체의 부분집합. """ state: Dict[Tuple[int, int], Optional[str]] = {} for s_idx, scene in enumerate(scene_detail_data.get("scenes", [])): for v_idx, v in enumerate(scene.get("t2i_variations", [])): state[(s_idx, v_idx)] = v.get("render_prompt_card_hash") return state def _assert_card_hash_unchanged( scene_detail_data: Dict, applied_indices: List[Tuple[int, int]], pre_state: Dict[Tuple[int, int], Optional[str]], ) -> None: """t2i_review 는 card payload 변경 금지 — pre/post hash 일치 검증 (AC-A3). V2 patch P2: pre_state 는 caller 가 _capture_card_hash_state(scene_detail_data) 로 mutation 전 만든 dict — 전체 매핑. 본 helper 는 applied_indices 만 비교. """ for s_idx, v_idx in applied_indices: v = scene_detail_data["scenes"][s_idx]["t2i_variations"][v_idx] post_hash = v.get("render_prompt_card_hash") if pre_state.get((s_idx, v_idx)) != post_hash: raise AppError( code="t2i_review.card_hash_unexpected_change", message=( f"t2i_review 가 card hash 를 변경함 — scope 외 mutation: " f"scenes[{s_idx}].t2i_variations[{v_idx}]" ), ) ``` `T2iReviewStep._execute()` 의 `run_t2i_review` 호출 부분 (line ~50 부근) 을 다음으로 재작성 — **mutation 호출까지 포함**: ```python # V2 patch P2 (Codex BLOCKING #2 + spec V5 S1): # pre-capture 는 run_t2i_review **호출 전** 의무 — review 가 내부에서 _apply_*_fixes 로 # scene_detail_data 를 mutate (backend/app/modules/pipeline/t2i_review.py:60-65). 호출 후 # capture 는 post-mutation state 라 assert 무의미. pre_card_state = _capture_card_hash_state(scene_detail_data) review_result = run_t2i_review( entity_t2i_data=entity_t2i_data, scene_detail_data=scene_detail_data, entity_merge_data=entity_merge_data, entity_detail_data=entity_detail_data, shot_extract_data=shot_extract_data, vwr_data=vwr_data, shot_staging_data=shot_staging_data, opik_metadata=opik_metadata, ) # 이 시점에 scene_detail_data 는 mutation 끝남 — pre_card_state 는 mutation 전 보존됨. # 수정된 데이터를 entity_t2i, scene_detail 체크포인트에 덮어쓰기 if review_result["entity_applied"] > 0: self._save_checkpoint_data("entity_t2i", entity_t2i_data) logger.info("t2i_review: entity_t2i checkpoint updated (%d fixes)", review_result["entity_applied"]) if review_result["scene_applied"] > 0: # AC-A1, A2, A3: sentinel refresh + card hash assert (Block A hotfix) applied_indices = review_result["scene_applied_indices"] try: refreshed = _refresh_scene_detail_sentinels(scene_detail_data, applied_indices) _assert_card_hash_unchanged(scene_detail_data, applied_indices, pre_card_state) except (AppError, KeyError, IndexError) as exc: # AC-A2: refresh / assert 실패 → t2i_review failed (silent corruption 차단) if isinstance(exc, AppError): raise raise AppError( code="t2i_review.sentinel_refresh_failed", message=f"sentinel refresh 실패: {exc}", ) from exc self._save_checkpoint_data("scene_detail", scene_detail_data) logger.info( "t2i_review: scene_detail checkpoint updated (%d fixes, %d sentinels refreshed)", review_result["scene_applied"], refreshed, ) ``` **P2 검증 의무 (Step 4 단위 테스트 추가)**: `test_t2i_review_sentinel_refresh.py` 에 추가 — pre-capture timing 회귀 가드: ```python def test_pre_card_state_captured_before_mutation(): """V2 patch P2 (Codex BLOCKING #2): pre_card_state 는 mutation 전 capture 되어야. run_t2i_review 가 scene_detail_data 를 mutate 하는 monkey 를 깔아도 _capture_card_hash_state 결과가 mutation 전 값이어야 assert 통과. """ from app.core.steps.t2i_review_step import _capture_card_hash_state scene_data = {"scenes": [{"t2i_variations": [ {"render_prompt_card_hash": "PRE_HASH"}, ]}]} pre = _capture_card_hash_state(scene_data) # 가짜 mutation scene_data["scenes"][0]["t2i_variations"][0]["render_prompt_card_hash"] = "POST_HASH" # pre 는 mutation 전 값으로 freeze assert pre[(0, 0)] == "PRE_HASH" ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_sentinel_refresh.py -v` Expected: 4 passed (V2.1 patch — Codex IMPORTANT #2: V2 P2 의 `test_pre_card_state_captured_before_mutation` 1건 추가 반영) - [ ] **Step 5: commit** ```bash git add backend/app/core/steps/t2i_review_step.py backend/tests/unit/test_t2i_review_sentinel_refresh.py git commit -m "feat(t2i_review_step): sentinel refresh + card hash assert (Block A AC-A1..A5)" ``` --- ### Task A7: Block A 통합 회귀 — fix 후 next resume 시 verify 통과 **Files:** - Test: `backend/tests/unit/test_t2i_review_sentinel_refresh.py` (확장) **AC:** AC-A4 - [ ] **Step 1: 통합 테스트 추가** ```python # backend/tests/unit/test_t2i_review_sentinel_refresh.py 에 append def test_full_cycle_refresh_then_verify_passes(): """AC-A4: t2i_review fix → sentinel refresh 후 scene_detail.verify_completion drift 0.""" from app.core.steps._owned_helpers import ( compute_t2i_prompt_hash, compute_owned_hash, compute_camera_direction_hash, OWNED_VALIDATOR_FULL, OWNED_SENTINEL_SCHEMA_VERSION, ) # 1) scene_detail 초기 상태 — sentinel 이 t2i_prompt 와 일치 scene_data = {"scenes": [{ "t2i_variations": [{ "t2i_prompt": "ORIGINAL prompt text", "owned_validation": { "schema_version": OWNED_SENTINEL_SCHEMA_VERSION, "t2i_prompt_hash": compute_t2i_prompt_hash("ORIGINAL prompt text"), "owned_hash": compute_owned_hash(["C01"]), "camera_direction_hash": compute_camera_direction_hash("static"), "validator": OWNED_VALIDATOR_FULL, "violations": [], }, }] }]} # 2) t2i_review 가 t2i_prompt 변경 (in-place) scene_data["scenes"][0]["t2i_variations"][0]["t2i_prompt"] = "REVISED prompt text" # 3) sentinel refresh 호출 refreshed = _refresh_scene_detail_sentinels(scene_data, applied_indices=[(0, 0)]) assert refreshed == 1 # 4) post-refresh sentinel 이 새 t2i_prompt 와 일치 → drift 0 sentinel = scene_data["scenes"][0]["t2i_variations"][0]["owned_validation"] assert sentinel["t2i_prompt_hash"] == compute_t2i_prompt_hash("REVISED prompt text") ``` - [ ] **Step 2: 테스트 실행** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_t2i_review_sentinel_refresh.py -v` Expected: 4 passed - [ ] **Step 3: 전체 회귀 — 기존 t2i_review 관련 테스트가 깨지지 않았는지** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k t2i_review -v` Expected: 모든 기존 + 신규 테스트 PASS - [ ] **Step 4: Block A 광역 smoke test** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_owned_helpers_refresh.py tests/unit/test_t2i_review_item_id_map.py tests/unit/test_t2i_review_apply_scene_fixes.py tests/unit/test_t2i_review_sentinel_refresh.py -v` Expected: 모두 PASS - [ ] **Step 5: commit** ```bash git add backend/tests/unit/test_t2i_review_sentinel_refresh.py git commit -m "test(t2i_review): full cycle refresh→verify integration (AC-A4)" ``` --- ## Block B — StepRunner decision 정리 (~1d) ### Task B0: `_get_step_run()` dict shape 변경 + 기존 caller 갱신 (V2 patch P4 + spec V5 S7) **V2 patch P4 (Codex BLOCKING #4 — `_get_step_run` tuple shape 오해석)**: 현 `_get_step_run()` 은 tuple `(status, completed_count, applicable_count)` 반환 (`backend/app/core/step_runner.py:465-471`). spec V5 S7 의 dict shape `{status, run_id, started_at, completed_count, applicable_count, recovery_count, updated_at}` 로 변경 — atomic claim / stale steal / verify path 가 `run_id` + `started_at` 의존이라 tuple 불충분. 본 task 는 Block B/C 의 모든 후속 task 의 prerequisite. **B1 진입 전 반드시 완료**. **Files:** - Modify: `backend/app/core/step_runner.py:465-471` (`_get_step_run` 본문 + SQL 컬럼 명시 SELECT) - Modify: `backend/app/core/step_runner.py:534-535` (resume 분기의 `existing[0] == "completed"` → `existing["status"] == "completed"`) - Modify: `backend/app/core/step_runner.py` 의 다른 caller (있다면 grep 으로 식별) - Test: `backend/tests/unit/test_get_step_run_shape.py` (신규) **AC:** AC-B5 / AC-C1 / AC-C8 prerequisite (Block C 의 atomic claim 이 새 shape 의존) - [ ] **Step 1: 기존 caller 식별 (grep)** ```bash # tuple-index 접근 패턴 + 변수 binding 모두 검색 rg "_get_step_run\(" backend/app/ backend/tests/ # 호출 site rg "existing\[\d\]" backend/app/core/step_runner.py # tuple-index 접근 (있을 가능성) ``` 기대: 1) `step_runner.py:534-535` 의 `existing[0] == "completed"` 1건 (이번 task 에서 변환) 2) 다른 file 에서 `_get_step_run` 호출 0건 (run() 내부 한정) 3) tests 의 fixture (있다면 함께 갱신) - [ ] **Step 2: failing test 작성** ```python # backend/tests/unit/test_get_step_run_shape.py """_get_step_run() dict shape 회귀 — V2 patch P4 + spec V5 S7.""" from unittest.mock import MagicMock from app.core.step_runner import StepRunner def _make_runner_with_row(row_obj) -> StepRunner: runner = StepRunner.__new__(StepRunner) runner.project_id = "p" runner.episode_id = "e" db = MagicMock() fetched = MagicMock() fetched.fetchone = MagicMock(return_value=row_obj) db.execute = MagicMock(return_value=fetched) runner.db = db return runner def test_get_step_run_returns_dict_with_required_keys(): """V5 S7: dict 반환 + 필수 키 모두 포함.""" row = MagicMock() row.status = "completed" row.run_id = "r1" row.started_at = "2026-05-08T00:00:00+00:00" row.completed_count = 5 row.applicable_count = 5 row.recovery_count = 0 row.updated_at = "2026-05-08T00:01:00+00:00" runner = _make_runner_with_row(row) result = runner._get_step_run("scene_detail") assert isinstance(result, dict) assert set(result.keys()) >= { "status", "run_id", "started_at", "completed_count", "applicable_count", "recovery_count", "updated_at", } assert result["status"] == "completed" assert result["run_id"] == "r1" assert result["started_at"] == "2026-05-08T00:00:00+00:00" def test_get_step_run_returns_none_when_no_row(): runner = _make_runner_with_row(None) assert runner._get_step_run("scene_detail") is None ``` Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_get_step_run_shape.py -v` Expected: FAIL — 현재 tuple 반환 - [ ] **Step 3: `_get_step_run()` dict 반환으로 변경** `backend/app/core/step_runner.py:465-471` 교체: ```python def _get_step_run(self, step_id: str) -> Optional[Dict[str, Any]]: """step_run row 의 모든 결정 필드를 dict 로 반환. V2 patch P4 + spec V5 S7 (Codex BLOCKING #4): atomic claim / stale steal / verify path 가 run_id + started_at + recovery_count 의존 — tuple index 오해석 위험 차단. """ row = self.db.execute(text(""" SELECT status, run_id, started_at, completed_count, applicable_count, COALESCE(recovery_count, 0) AS recovery_count, updated_at FROM step_run WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid """), {"pid": self.project_id, "eid": self.episode_id, "sid": step_id}).fetchone() if row is None: return None return { "status": row.status, "run_id": row.run_id, "started_at": row.started_at, "completed_count": row.completed_count, "applicable_count": row.applicable_count, "recovery_count": row.recovery_count, "updated_at": row.updated_at, } ``` - [ ] **Step 4: 기존 caller 갱신** `step_runner.py:534-535` 의 resume 분기: ```python # 기존: # existing = self._get_step_run(self.step_id) # if existing and existing[0] == "completed": # 새로 (V2 patch P4): existing = self._get_step_run(self.step_id) if existing and existing["status"] == "completed": ``` 다른 caller (Step 1 grep 결과 따라) 도 동일 패턴으로 dict key 접근 변환. - [ ] **Step 5: 테스트 + 회귀 검증** ```bash cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_get_step_run_shape.py -v # Expected: 2 passed # Block A baseline 회귀 cd backend && PYTHONPATH=. .venv/bin/pytest tests/ -v -k "step_runner or step_execution" # Expected: 모두 PASS (회귀 0) # tuple-index 잔존 grep — 0건 (또는 명시 fallback 코멘트) rg "_get_step_run\([^)]*\)\s*\[\d" backend/app/ backend/tests/ # → 0 lines rg "existing\[\d\]" backend/app/core/step_runner.py # → 0 lines ``` - [ ] **Step 6: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_get_step_run_shape.py git commit -m "refactor(step_runner): _get_step_run dict shape (V5 S7 / Codex NEEDS_REVISION BLOCKING #4)" ``` --- ### Task B1: `CompletionReport.origin` 필드 추가 **Files:** - Modify: `backend/app/core/integrity_report.py` - Test: `backend/tests/unit/test_completion_report_origin.py` (신규) **AC:** AC-B9 (V2 patch I2 backward compat 기반) - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_completion_report_origin.py """CompletionReport.origin 필드 — backward-compatible default.""" from app.core.integrity_report import CompletionReport def test_origin_default_is_artifact_missing(): """기존 verifier 가 origin 미지정 시 default 'artifact_missing'.""" report = CompletionReport(is_complete=False, missing=["x"], severity="missing", metadata={}) assert report.origin == "artifact_missing" def test_origin_can_be_overridden(): """verifier 가 origin 명시 가능.""" report = CompletionReport( is_complete=False, missing=["sentinel"], severity="partial", metadata={}, origin="invariant_drift", ) assert report.origin == "invariant_drift" def test_origin_clean_when_complete(): """is_complete=True 시 origin='clean' 권장 (자유).""" report = CompletionReport( is_complete=True, missing=[], severity="clean", metadata={}, origin="clean", ) assert report.is_complete is True assert report.origin == "clean" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_completion_report_origin.py -v` Expected: FAIL — `origin` field 없음 - [ ] **Step 3: `CompletionReport` 확장** `backend/app/core/integrity_report.py`: ```python """Step 산출물 무결성 검증 리포트 dataclass.""" from dataclasses import dataclass, field from typing import Literal @dataclass(frozen=True) class CompletionReport: """Step 산출물 무결성 검증 결과. is_complete=True → resume entry verify 에서 skipped 결정 + exit verify 에서 completed 마킹. is_complete=False → entry: ResumeDecision evaluation, exit: status='partial' 마킹. origin (V2 patch I1, spec §2.1): verify 결과의 분류. ResumeDecision 평가 시 사용. - clean: 정상 - artifact_missing: DB row / PNG / cp 파일 부재 (default — backward-compat) - contract_drift: schema/config_hash mismatch / loader contract / AppError - invariant_drift: sentinel/hash drift (mutator/manual/unknown) - verify_crashed: verifier 의 unexpected exception """ is_complete: bool missing: list[str] severity: Literal["clean", "partial", "missing"] metadata: dict origin: Literal[ "clean", "artifact_missing", "contract_drift", "invariant_drift", "verify_crashed", ] = "artifact_missing" @dataclass(frozen=True) class CleanupReport: """force/auto-rerun 시 stale artifact 정리 결과.""" deleted_db_rows: int deleted_files: int targets: list[str] skipped: list[str] ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_completion_report_origin.py -v` Expected: 3 passed - [ ] **Step 5: 기존 verifier 회귀 — origin default 작동** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k completion_report -v && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k integrity -v` Expected: 기존 + 신규 모두 PASS - [ ] **Step 6: commit** ```bash git add backend/app/core/integrity_report.py backend/tests/unit/test_completion_report_origin.py git commit -m "feat(integrity_report): CompletionReport.origin field with backward-compat default (AC-B9)" ``` --- ### Task B2: `settings.step_running_timeout_seconds` 추가 **Files:** - Modify: `backend/app/core/config.py` - Test: `backend/tests/unit/test_config_running_timeout.py` (신규) **AC:** AC-C12 (V3 patch I3) - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_config_running_timeout.py """step_running_timeout_seconds config — AC-C12 (V3 patch I3).""" from app.core.config import settings def test_default_is_3600_seconds(): assert settings.step_running_timeout_seconds == 3600 def test_can_override_via_env(monkeypatch): monkeypatch.setenv("STEP_RUNNING_TIMEOUT_SECONDS", "7200") from app.core.config import Settings s = Settings() assert s.step_running_timeout_seconds == 7200 ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_config_running_timeout.py -v` Expected: FAIL — `step_running_timeout_seconds` attribute 없음 - [ ] **Step 3: config 추가** `backend/app/core/config.py` 의 `Settings` class 안 적절한 위치 (timeout 관련 setting 그룹 근처) 에 추가: ```python class Settings(BaseSettings): # ... 기존 ... # Resume architecture fix (Block B/C, V3 patch I3) step_running_timeout_seconds: int = 3600 """running status row 가 stale 로 분류되는 elapsed 임계값 (초). `_evaluate_running_state()` 가 started_at 으로부터 이 값을 초과한 row 를 STALE_RUNNING_RECOVERY 로 분류 → atomic claim 으로 expected-match steal 시도. """ ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_config_running_timeout.py -v` Expected: 2 passed - [ ] **Step 5: commit** ```bash git add backend/app/core/config.py backend/tests/unit/test_config_running_timeout.py git commit -m "feat(config): step_running_timeout_seconds with default 3600 (AC-C12)" ``` --- ### Task B3: `ResumeAction` enum + `ResumeDecision` dataclass **Files:** - Modify: `backend/app/core/step_runner.py` (상단 imports + class 정의) - Test: `backend/tests/unit/test_resume_decision.py` (신규) **AC:** AC-B1 (V2 patch I1) - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_resume_decision.py """ResumeAction + ResumeDecision — AC-B1 (V2 patch I1).""" from app.core.step_runner import ResumeAction, ResumeDecision def test_action_enum_has_six_values(): assert {a.value for a in ResumeAction} == { "skip", "rerun_self", "force_explicit", "stale_running_recovery", "block", "not_applicable", } def test_decision_default_optional_fields(): d = ResumeDecision(action=ResumeAction.SKIP, reason="cp clean") assert d.origin is None assert d.expected_started_at is None assert d.expected_run_id is None def test_decision_with_expected_fields(): d = ResumeDecision( action=ResumeAction.STALE_RUNNING_RECOVERY, reason="elapsed 4000s > 3600s", expected_started_at="2026-05-07T10:00:00+00:00", expected_run_id="run-abc", ) assert d.action == ResumeAction.STALE_RUNNING_RECOVERY assert d.expected_started_at == "2026-05-07T10:00:00+00:00" assert d.expected_run_id == "run-abc" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: FAIL — `cannot import name 'ResumeAction'` - [ ] **Step 3: enum + dataclass 정의** `backend/app/core/step_runner.py` 상단 (logger 정의 직후) 에 추가: ```python from dataclasses import dataclass from enum import Enum from typing import Optional class ResumeAction(Enum): """ResumeDecision.action — claim 후 실행 분기 결정 (V2 patch I1, spec §2.3).""" SKIP = "skip" RERUN_SELF = "rerun_self" FORCE_EXPLICIT = "force_explicit" STALE_RUNNING_RECOVERY = "stale_running_recovery" BLOCK = "block" NOT_APPLICABLE = "not_applicable" @dataclass(frozen=True) class ResumeDecision: """판정 결과 + 사유. V3 patch B1: STALE_RUNNING_RECOVERY 시 expected_started_at + expected_run_id 동반. claim SQL 의 atomic steal 조건에서 SQL cast 없이 (text 비교만) 검증. """ action: ResumeAction reason: str origin: Optional[str] = None # CompletionReport.origin 매핑 expected_started_at: Optional[str] = None expected_run_id: Optional[str] = None ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: 3 passed - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_resume_decision.py git commit -m "feat(step_runner): ResumeAction enum + ResumeDecision dataclass (AC-B1)" ``` --- ### Task B4: `_evaluate_running_state()` — running timeout **Files:** - Modify: `backend/app/core/step_runner.py` (`_evaluate_running_state` 신규 메서드) - Test: `backend/tests/unit/test_running_state_evaluation.py` (신규) **AC:** AC-B5 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_running_state_evaluation.py """_evaluate_running_state — AC-B5.""" from datetime import datetime, timezone, timedelta from unittest.mock import MagicMock import pytest from app.core.step_runner import ResumeAction, StepRunner def _make_runner(monkeypatch=None): runner = StepRunner.__new__(StepRunner) runner.step_id = "test_step" return runner def test_started_at_null_returns_block(): """started_at NULL → BLOCK (자동 rerun 금지).""" runner = _make_runner() decision = runner._evaluate_running_state({"started_at": None, "run_id": "r1"}) assert decision.action == ResumeAction.BLOCK assert "NULL" in decision.reason or "None" in decision.reason def test_started_at_parse_failed_returns_block(): runner = _make_runner() decision = runner._evaluate_running_state({ "started_at": "not-a-timestamp", "run_id": "r1", }) assert decision.action == ResumeAction.BLOCK assert "parse" in decision.reason.lower() def test_within_timeout_returns_block(monkeypatch): runner = _make_runner() recent = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat() decision = runner._evaluate_running_state({ "started_at": recent, "run_id": "r1", }) assert decision.action == ResumeAction.BLOCK assert "healthy" in decision.reason.lower() def test_over_timeout_returns_stale_recovery_with_expected_fields(monkeypatch): runner = _make_runner() long_ago = (datetime.now(timezone.utc) - timedelta(seconds=5000)).isoformat() decision = runner._evaluate_running_state({ "started_at": long_ago, "run_id": "r1", }) assert decision.action == ResumeAction.STALE_RUNNING_RECOVERY # V3 patch B1: expected_* 필드 채워야 함 assert decision.expected_started_at == long_ago assert decision.expected_run_id == "r1" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_running_state_evaluation.py -v` Expected: FAIL — `_evaluate_running_state` 미정의 - [ ] **Step 3: 메서드 구현** `backend/app/core/step_runner.py` `StepRunner` class 안에 추가: ```python def _evaluate_running_state(self, existing: dict) -> "ResumeDecision": """running status row 평가. AC-B5 (spec §4.6). V3 patch B1: STALE_RUNNING_RECOVERY 시 expected_started_at + expected_run_id 채움. """ from datetime import datetime, timezone from app.core.config import settings started_at_raw = existing.get("started_at") run_id_raw = existing.get("run_id") if started_at_raw is None: logger.warning( "[STALE_RUNNING] step=%s started_at=NULL → BLOCK", self.step_id, ) return ResumeDecision( action=ResumeAction.BLOCK, reason="running with started_at=NULL (manual investigation required)", ) try: started_at = datetime.fromisoformat(started_at_raw.replace("Z", "+00:00")) except (ValueError, TypeError, AttributeError) as exc: logger.warning( "[STALE_RUNNING] step=%s started_at parse failed (%r): %s → BLOCK", self.step_id, started_at_raw, exc, ) return ResumeDecision( action=ResumeAction.BLOCK, reason=f"running with started_at parse failed: {started_at_raw!r}", ) 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)", ) logger.warning( "[STALE_RUNNING] step=%s elapsed=%.0fs timeout=%ds → STALE_RUNNING_RECOVERY", self.step_id, elapsed, timeout_seconds, ) return ResumeDecision( action=ResumeAction.STALE_RUNNING_RECOVERY, reason=f"running stale (elapsed={elapsed:.0f}s > timeout={timeout_seconds}s)", expected_started_at=started_at_raw, expected_run_id=run_id_raw, ) ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_running_state_evaluation.py -v` Expected: 4 passed - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_running_state_evaluation.py git commit -m "feat(step_runner): _evaluate_running_state with timeout-based STALE_RUNNING_RECOVERY (AC-B5)" ``` --- ### Task B5: `_LEGACY_SCHEMA_BUMP_ALLOWLIST` + `_evaluate_contract_drift()` **Files:** - Modify: `backend/app/core/step_manifest.py` (allowlist 추가) - Modify: `backend/app/core/step_runner.py` (`_evaluate_contract_drift`) - Test: `backend/tests/unit/test_legacy_schema_bump_allowlist.py` (신규) **AC:** AC-B4 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_legacy_schema_bump_allowlist.py """_LEGACY_SCHEMA_BUMP_ALLOWLIST — AC-B4.""" from app.core.step_manifest import _LEGACY_SCHEMA_BUMP_ALLOWLIST def test_allowlist_contains_only_entity_t2i(): """V1 patch I3 + V2 질문 답변 #3: scene_detail 등 추가 차단.""" assert _LEGACY_SCHEMA_BUMP_ALLOWLIST == frozenset({"entity_t2i"}) def test_scene_detail_not_in_allowlist(): """rerun_self 시 downstream 보존 → drift 위험으로 제외.""" assert "scene_detail" not in _LEGACY_SCHEMA_BUMP_ALLOWLIST def test_evaluate_contract_drift_allowlist_returns_rerun_self(): """allowlist 내 step + schema_version mismatch → RERUN_SELF.""" from app.core.step_runner import ResumeAction, StepRunner runner = StepRunner.__new__(StepRunner) runner.step_id = "entity_t2i" decision = runner._evaluate_contract_drift("schema_version mismatch: 체크포인트=1, 현재=2") assert decision.action == ResumeAction.RERUN_SELF def test_evaluate_contract_drift_default_blocks(): from app.core.step_runner import ResumeAction, StepRunner runner = StepRunner.__new__(StepRunner) runner.step_id = "scene_detail" decision = runner._evaluate_contract_drift("schema_version mismatch: ...") assert decision.action == ResumeAction.BLOCK def test_evaluate_contract_drift_non_schema_blocks(): from app.core.step_runner import ResumeAction, StepRunner runner = StepRunner.__new__(StepRunner) runner.step_id = "entity_t2i" decision = runner._evaluate_contract_drift("config_hash mismatch: ...") # config_hash 는 allowlist 적용 대상 아님 → BLOCK assert decision.action == ResumeAction.BLOCK ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_legacy_schema_bump_allowlist.py -v` Expected: FAIL — `_LEGACY_SCHEMA_BUMP_ALLOWLIST` 미정의 - [ ] **Step 3: allowlist + 메서드 구현** `backend/app/core/step_manifest.py` 파일 적절한 위치 (export 가능한 module-level constant) 에 추가: ```python # V1 patch I3 + V2 질문 답변 #3 (spec §4.5): # 한정적 — entity_t2i 만. scene_detail 제거 사유: rerun_self 시 downstream # (shot_dependency_t2i / scene_image_pipeline 등) 보존이라 새 scene_detail 결과와 # 기존 downstream 사이 불일치 위험. scene_detail 은 사용자 명시 force 또는 # D3 manifest flag 도입 후 허용. # expiry: D3 deploy 시점에 제거 (manifest.allow_auto_rerun_on_schema_bump 로 대체). _LEGACY_SCHEMA_BUMP_ALLOWLIST = frozenset({ "entity_t2i", }) ``` `backend/app/core/step_runner.py` `StepRunner` class 안에 추가: ```python def _evaluate_contract_drift(self, mismatch_reason: str) -> "ResumeDecision": """contract_drift 분류 — schema/config_hash mismatch. AC-B4 (spec §4.5): legacy schema bump allowlist (entity_t2i 만) 임시 적용. 그 외는 BLOCK (사용자 명시 force 요구). """ from app.core.step_manifest import _LEGACY_SCHEMA_BUMP_ALLOWLIST 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} ({mismatch_reason})", origin="contract_drift", ) return ResumeDecision( action=ResumeAction.BLOCK, reason=mismatch_reason, origin="contract_drift", ) ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_legacy_schema_bump_allowlist.py -v` Expected: 5 passed - [ ] **Step 5: commit** ```bash git add backend/app/core/step_manifest.py backend/app/core/step_runner.py backend/tests/unit/test_legacy_schema_bump_allowlist.py git commit -m "feat(step_runner): _LEGACY_SCHEMA_BUMP_ALLOWLIST + _evaluate_contract_drift (AC-B4)" ``` --- ### Task B6: `_evaluate_invariant_drift()` — 1차 BLOCK 정책 **Files:** - Modify: `backend/app/core/step_runner.py` - Test: `backend/tests/unit/test_resume_decision.py` (확장) **AC:** AC-B3 - [ ] **Step 1: 테스트 추가** `backend/tests/unit/test_resume_decision.py` 에 append: ```python def test_evaluate_invariant_drift_blocks(): """AC-B3: 1차 정책 — invariant_drift 무조건 BLOCK.""" from app.core.step_runner import ResumeAction, StepRunner from app.core.integrity_report import CompletionReport runner = StepRunner.__new__(StepRunner) runner.step_id = "scene_detail" report = CompletionReport( is_complete=False, missing=["3 owned_validation sentinel_drifted: ..."], severity="missing", metadata={}, origin="invariant_drift", ) decision = runner._evaluate_invariant_drift(report) assert decision.action == ResumeAction.BLOCK assert decision.origin == "invariant_drift" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py::test_evaluate_invariant_drift_blocks -v` Expected: FAIL - [ ] **Step 3: 메서드 구현** `backend/app/core/step_runner.py` `StepRunner` 안: ```python def _evaluate_invariant_drift(self, report) -> "ResumeDecision": """1차 정책 (D4 까지): invariant_drift 무조건 BLOCK. AC-B3 (spec §4.4): mutator origin 식별 메커니즘 D4 도입 전까지 unknown 분류 → safer default BLOCK. 사용자 명시 force 시 진행. """ return ResumeDecision( action=ResumeAction.BLOCK, reason=f"invariant_drift (D4 까지 임시 BLOCK 정책): {report.missing[:3]}", origin="invariant_drift", ) ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: 모두 PASS - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_resume_decision.py git commit -m "feat(step_runner): _evaluate_invariant_drift (1차 BLOCK 정책, AC-B3)" ``` --- ### Task B7: `_safe_verify_completion` — verify_crashed fail-fast (entry/exit 양쪽) **Files:** - Modify: `backend/app/core/step_runner.py` - Test: `backend/tests/unit/test_step_runner_verify_crashed.py` (신규) **AC:** AC-B2, AC-B8 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_step_runner_verify_crashed.py """verify_crashed fail-fast — AC-B2, AC-B8.""" import pytest from app.core.errors import AppError from app.core.step_runner import StepRunner from app.core.integrity_report import CompletionReport def _make_runner(verify_fn): runner = StepRunner.__new__(StepRunner) runner.step_id = "test_step" runner.verify_completion = verify_fn return runner def test_apperror_returns_contract_drift(): """AppError 계열 (e.g. step.contract_violation) → CompletionReport(origin=contract_drift).""" def verify(): raise AppError(code="step.contract_violation", message="loader violation") runner = _make_runner(verify) report = runner._safe_verify_completion() assert report.is_complete is False assert report.origin == "contract_drift" assert "contract" in (report.missing[0] if report.missing else "").lower() \ or "loader violation" in (report.missing[0] if report.missing else "") def test_unexpected_exception_raises_verify_crashed(): """unexpected exception (KeyError 등) → AppError(verify_crashed) raise.""" def verify(): raise KeyError("missing key 'foo'") runner = _make_runner(verify) with pytest.raises(AppError) as exc: runner._safe_verify_completion() assert exc.value.code == "step.verify_crashed" def test_no_exception_returns_completion_report(): def verify(): return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={}, origin="clean", ) runner = _make_runner(verify) report = runner._safe_verify_completion() assert report.is_complete is True ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_step_runner_verify_crashed.py -v` Expected: FAIL — 현재 `_safe_verify_completion` 이 모든 예외를 `is_complete=False` 로 변환 - [ ] **Step 3: `_safe_verify_completion` 변경** `backend/app/core/step_runner.py` 의 `_safe_verify_completion` (line ~842) 을 다음으로 교체: ```python def _safe_verify_completion(self) -> "CompletionReport": """verify_completion 실행 + 예외 분류. AC-B2, AC-B8 (spec §4.3): - AppError 계열 (loader contract / step.contract_violation 등) → CompletionReport with origin='contract_drift' (caller 가 BLOCK 처리) - unexpected exception (KeyError, ValueError, AttributeError 등) → AppError(step.verify_crashed) raise (자동 force 금지, fail-fast) """ from app.core.errors import AppError from app.core.integrity_report import CompletionReport try: return self.verify_completion() except AppError as exc: # contract 계열 — caller 가 BLOCK 처리 logger.warning( "verify_completion AppError for %s: %s (code=%s)", self.step_id, exc, getattr(exc, "code", "?"), ) return CompletionReport( is_complete=False, missing=[f"contract: {exc}"], 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 ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_step_runner_verify_crashed.py -v` Expected: 3 passed - [ ] **Step 5: 광범위 회귀 — 기존 verify_completion override 가 깨지지 않았는지** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k verify -v` Expected: 기존 테스트 + 신규 테스트 모두 PASS - [ ] **Step 6: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_step_runner_verify_crashed.py git commit -m "feat(step_runner): _safe_verify_completion fail-fast on unexpected exception (AC-B2, B8)" ``` --- ### Task B8: scene_detail.verify_completion 의 origin 분류 (V2 patch P5 + spec V5 S4) **V2 patch P5 (Codex BLOCKING #6 + spec V5 S4)**: 이전 v1 plan 의 단위 테스트는 `CompletionReport(origin=...)` 객체를 직접 build 하고 assert 만 했기 때문에 `SceneDetailStep.verify_completion()` 안의 분기 (loader_violations / sentinel / card / unexpected exception 우선순위 + severity 분류) 를 검증하지 않았음. **본 task 의 단위 테스트는 `verify_completion()` 을 fixture 기반으로 직접 호출해 7 branch 각각의 origin/severity 분류를 검증** 의무 (spec §4.3 V5 S4 표 참조). **Files:** - Modify: `backend/app/core/steps/detail_steps.py` (`verify_completion` 본문, line ~1133/~1276/~1373/~1404 부근) - Test: `backend/tests/unit/test_scene_detail_verify_origin.py` (신규, **fixture 기반**) **AC:** AC-B9 - [ ] **Step 1: fixture builder + failing tests 작성** ```python # backend/tests/unit/test_scene_detail_verify_origin.py """scene_detail.verify_completion origin 분류 — AC-B9 (V5 S4 / V2 patch P5). V5 S4 (Codex BLOCKING #6): SceneDetailStep.verify_completion() 의 실제 분기를 fixture 로 직접 태움. CompletionReport(origin=...) 객체를 직접 build 하는 helper-only 테스트는 loader_violations / sentinel / card 우선순위 + severity (missing/partial) 분류 매핑 미검증이라 제외. Fixture 설계 가이드: _scene_detail_cp_full() 는 detail_steps.py:1068 (verify_completion 시작) 부터 읽어, 모든 sentinel hash 가 t2i_prompt 와 일치하는 완전 cp 를 build. 구현자가 실제 verify 코드를 보고 필드 shape (owned_validation / render_prompt_card / sentinel hash 함수) 를 따라야 함. """ import pytest from unittest.mock import MagicMock, patch from app.core.steps.detail_steps import SceneDetailStep from app.core.errors import AppError @pytest.fixture def make_step(): """SceneDetailStep 인스턴스 (생성자 우회) — verify_completion 만 호출 가능. 구현자: SceneDetailStep.__init__ 가 의존하는 attribute 만 set. project_id/episode_id/ step_id + load_checkpoint stub 정도면 verify_completion 호출 가능. """ def _make(scene_data: dict, prev_cp: dict = None): step = SceneDetailStep.__new__(SceneDetailStep) step.project_id = "p" step.episode_id = "e" step.step_id = "scene_detail" step.load_checkpoint = MagicMock(return_value=scene_data) step._load_prev_checkpoint = MagicMock(return_value=prev_cp or {}) return step return _make def _scene_detail_cp_full() -> dict: """완전한 (drift 없는) scene_detail cp fixture. 구현자: detail_steps.py:1068 부터 verify_completion 본문을 읽고 모든 필드 채움 — scenes[].t2i_variations[].t2i_prompt + owned_validation (schema_version, t2i_prompt_hash, owned_hash, camera_direction_hash, validator) + render_prompt_card + render_prompt_card_hash. sentinel hash 는 compute_t2i_prompt_hash() 등으로 정확 계산해 일치시킴. """ raise NotImplementedError("구현자: detail_steps.py:1068+ 의 verify_completion 본문 read 후 작성") # === V5 S4 표 7 branch 각각 fixture 기반 검증 === # V2.1 patch (Codex BLOCKING #3): patch target 정정. # 실제 코드 (detail_steps.py:1129-1145) 는 SceneContextLoader(self) 객체의 메서드 호출 — # step instance 의 메서드가 아님. 또한 card recompute 는 함수 내부 import: # from app.core.steps.render_prompt_card import build_render_prompt_card as _g41_build_card # 따라서 module-level alias 가 아니라 source module 의 함수를 patch. def test_loader_violation_returns_contract_drift(make_step, monkeypatch): """V5 S4 branch 1: loader (`SceneContextLoader._load_chain_bg_owned_by_shot`) AppError → contract_drift.""" from app.core.steps.scene_context_loader import SceneContextLoader step = make_step(_scene_detail_cp_full()) # V2.1 patch: SceneContextLoader 의 메서드를 patch (step instance 아님) def fake_loader(self): raise AppError(code="step.contract_violation", message="missing required key") monkeypatch.setattr(SceneContextLoader, "_load_chain_bg_owned_by_shot", fake_loader) report = step.verify_completion() assert report.origin == "contract_drift" assert report.is_complete is False assert any("contract violation" in m or "loader" in m for m in report.missing) def test_card_recompute_apperror_returns_contract_drift(make_step): """V5 S4 branch 2: build_render_prompt_card AppError → contract_drift.""" step = make_step(_scene_detail_cp_full()) # V2.1 patch: source module 의 함수를 patch (detail_steps 의 alias 아님 — 함수 내부 import). with patch("app.core.steps.render_prompt_card.build_render_prompt_card") as fake: fake.side_effect = AppError(code="step.contract_violation", message="ctx mismatch") report = step.verify_completion() assert report.origin == "contract_drift" def test_card_recompute_unexpected_exception_raises_verify_crashed(make_step): """V5 S4 branch 3: build_render_prompt_card KeyError 등 → step.verify_crashed raise (contract 아님).""" step = make_step(_scene_detail_cp_full()) with patch("app.core.steps.render_prompt_card.build_render_prompt_card") as fake: fake.side_effect = KeyError("ctx['some_key']") with pytest.raises(AppError) as exc_info: step.verify_completion() assert exc_info.value.code == "step.verify_crashed" def test_sentinel_t2i_hash_drift_returns_invariant_drift(make_step): """V5 S4 branch 4: stored sentinel.t2i_prompt_hash != computed → invariant_drift.""" cp = _scene_detail_cp_full() cp["scenes"][0]["t2i_variations"][0]["owned_validation"]["t2i_prompt_hash"] = "STALE_HASH" step = make_step(cp) report = step.verify_completion() assert report.origin == "invariant_drift" def test_card_hash_drift_returns_invariant_drift(make_step): """V5 S4 branch 5: stored render_prompt_card_hash != computed → invariant_drift.""" cp = _scene_detail_cp_full() cp["scenes"][0]["t2i_variations"][0]["render_prompt_card_hash"] = "STALE_CARD_HASH" step = make_step(cp) report = step.verify_completion() assert report.origin == "invariant_drift" def test_owned_validation_missing_returns_contract_drift(make_step): """V5 S4 branch 6: owned_validation key 부재 → contract_drift (구조 위반).""" cp = _scene_detail_cp_full() del cp["scenes"][0]["t2i_variations"][0]["owned_validation"] step = make_step(cp) report = step.verify_completion() assert report.origin == "contract_drift" def test_render_prompt_card_payload_missing_returns_contract_drift(make_step): """V5 S4 branch 7: render_prompt_card top-level field 부재 → contract_drift.""" cp = _scene_detail_cp_full() if "render_prompt_card" in cp["scenes"][0]["t2i_variations"][0]: del cp["scenes"][0]["t2i_variations"][0]["render_prompt_card"] step = make_step(cp) report = step.verify_completion() assert report.origin == "contract_drift" # === 우선순위 + severity 검증 === def test_priority_loader_violation_over_sentinel_drift(make_step, monkeypatch): """V5 S4: loader_violation + sentinel_drift 동시 발생 시 contract_drift 우선.""" from app.core.steps.scene_context_loader import SceneContextLoader cp = _scene_detail_cp_full() cp["scenes"][0]["t2i_variations"][0]["owned_validation"]["t2i_prompt_hash"] = "STALE" step = make_step(cp) # V2.1 patch: SceneContextLoader 의 메서드를 patch (step instance 아님) def fake_loader(self): raise AppError(code="step.contract_violation", message="bg loader failed") monkeypatch.setattr(SceneContextLoader, "_load_chain_bg_owned_by_shot", fake_loader) report = step.verify_completion() assert report.origin == "contract_drift" # loader 우선 def test_severity_partial_when_partial_card_drift(make_step): """V5 S4: card drift 일부 (전체 아님) → severity='partial'. detail_steps.py:1394-1402 의 _all_card_missing 판정 — 일부 drift 시 partial. """ cp = _scene_detail_cp_full() # variation 한 개만 drift, 나머지는 정상 — 전체 N/total 미만 cp["scenes"][0]["t2i_variations"][0]["render_prompt_card_hash"] = "STALE" step = make_step(cp) report = step.verify_completion() assert report.severity == "partial" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_scene_detail_verify_origin.py -v` Expected: FAIL — `verify_completion` 의 분기에 `origin=...` 이 아직 배정 안 됐거나, fixture 가 NotImplementedError raise. Step 3 에서 분류 코드 추가 후 PASS. - [ ] **Step 3: `scene_detail.verify_completion` 의 분류 명시 (V2.1 patch — aggregation 보존)** **V2.1 patch (Codex IMPORTANT #1)**: 현재 verify 코드 (`detail_steps.py:1358-1419`) 는 `loader_violations` / `failed_indices` / `contract_violations` / `sentinel_drifted` / `card_drifted` 를 모아 `missing_msgs.extend(...)` aggregation 후 severity 계산 (line 1397-1402). 이 구조를 그대로 보존하고 **origin 필드만** 추가 — early-return 패턴은 단일 branch 만 보고하므로 aggregation 정보 손실. 우선순위 결정도 기존 severity 조건 (loader_violations or len(failed_indices) >= total → "missing") 을 그대로 origin 우선순위로 mirror. **핵심 패턴**: aggregation 끝의 단일 `return CompletionReport(...)` 자리에 origin 만 결정해 추가: ```python # detail_steps.py:1404 부근 — missing_msgs 계산 후 단일 return 자리. # 기존 (대략): # severity = ("missing" if loader_violations or ... else "partial") # return CompletionReport(is_complete=False, missing=missing_msgs, severity=severity, metadata={...}) # 새로 (V2.1 patch + V5 S4): severity = ( "missing" if loader_violations or len(failed_indices) >= total or _all_card_missing else "partial" ) # V5 S4 origin priority — 기존 severity 결정 순서 mirror. # loader/contract 우선 (구조 위반) → invariant (stored vs computed) → 기본 contract. if loader_violations: origin = "contract_drift" # AppError 계열 + loader contract elif contract_violations: origin = "contract_drift" # status='contract_violation' 마커 elif _owned_validation_missing or _card_payload_missing: # 명시 helper 결과 origin = "contract_drift" # 부재 / shape 위반 elif sentinel_drifted or card_drifted: origin = "invariant_drift" # stored vs computed mismatch else: origin = "contract_drift" # default safer (failed_indices only 등) return CompletionReport( is_complete=False, missing=missing_msgs, severity=severity, metadata={ "total_results": total, "failed_count": ( len(failed_indices) + len(contract_violations) + len(sentinel_drifted) + len(card_drifted) ), "failed_indices": failed_indices, "contract_violations": contract_violations, "sentinel_drifted": sentinel_drifted, "card_drifted": card_drifted, "loader_violations": loader_violations, }, origin=origin, # V5 S4 + V2.1 patch — aggregation 보존, origin 만 추가 ) ``` 추가 의무 — verify_completion 안의 helper 호출 자리에서 unexpected exception 격상: ```python # detail_steps.py:1229-1233 — 함수 내부 import 직후. # 기존: recomputed_card = _g41_build_card(ctx, ...) — exception 시 그대로 propagate (verify_crashed 미마킹) # 새로 (V5 S4 branch 3 — verify_crashed): try: recomputed_card = _g41_build_card(ctx, ...) except AppError: # contract_drift — loader_violations 처럼 aggregation 에 추가 loader_violations.append(f"card_recompute contract violation: ...") recomputed_card = None # downstream skip except Exception as exc: # V5 S4 branch 3: unexpected exception → step.verify_crashed raise (자동 force 금지) raise AppError( code="step.verify_crashed", message=f"build_render_prompt_card crashed: {type(exc).__name__}: {exc}", ) from exc ``` `_load_chain_bg_owned_by_shot` 의 AppError 는 이미 현 코드 (line 1135-1142) 에서 `loader_violations.append(...)` 로 aggregation 에 잘 들어가고 있음 — origin 분류만 위쪽 단일 return 에서 처리. 추가 try/except 불필요. - [ ] **Step 4: 테스트 실행 (broader)** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_scene_detail_verify_origin.py tests/unit/ -k scene_detail -v` Expected: 모든 테스트 PASS (기존 scene_detail tests 회귀 0) - [ ] **Step 5: commit** ```bash git add backend/app/core/steps/detail_steps.py backend/tests/unit/test_scene_detail_verify_origin.py git commit -m "feat(scene_detail): verify_completion origin classification (AC-B9)" ``` --- ### Task B9: `_update_step_run(require_owner=True)` 도입 + exception path **Files:** - Modify: `backend/app/core/step_runner.py` (`_update_step_run` 시그니처 + caller 들) - Test: `backend/tests/unit/test_step_runner_owner_aware_update.py` (신규) **AC:** AC-C7 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_step_runner_owner_aware_update.py """require_owner=True 가 모든 transition path 에 적용 — AC-C7.""" from unittest.mock import MagicMock from app.core.step_runner import StepRunner def test_update_step_run_includes_run_id_when_require_owner(): """require_owner=True 시 SQL 의 WHERE 절에 run_id 포함.""" runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1" runner.episode_id = "e1" runner.run_id = "r-self" runner.db = MagicMock() runner._now = lambda: "2026-05-07T00:00:00+00:00" runner._update_step_run("completed", require_owner=True) # MagicMock 으로 호출된 SQL 검증 call_args = runner.db.execute.call_args sql_text = str(call_args[0][0]) bind_params = call_args[0][1] if len(call_args[0]) > 1 else call_args.kwargs.get("params", {}) assert "run_id" in sql_text # WHERE 에 run_id 조건 # bind_params 에 run_id 포함 assert any(k in str(bind_params) for k in ("run_id", "rid")) or "r-self" in str(bind_params) def test_update_step_run_omits_run_id_when_not_require_owner(): runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1" runner.episode_id = "e1" runner.run_id = "r-self" runner.db = MagicMock() runner._now = lambda: "2026-05-07T00:00:00+00:00" runner._update_step_run("running", require_owner=False) sql_text = str(runner.db.execute.call_args[0][0]) # WHERE 에 run_id 조건 없어야 함 (default 동작) # 정확한 검증은 SQL 구조 의존 — 간단히 require_owner=False 시 SQL 짧음 assert "AND run_id" not in sql_text ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_step_runner_owner_aware_update.py -v` Expected: FAIL — 시그니처 mismatch - [ ] **Step 3: `_update_step_run` 시그니처 변경** `backend/app/core/step_runner.py` 의 `_update_step_run` (현재 line ~480-510 부근) 시그니처에 `require_owner: bool = False` 추가, SQL 의 WHERE 절 동적 구성: ```python def _update_step_run( self, status: str, *, require_owner: bool = False, completed_count: int = 0, applicable_count: int = 1, failed_count: int = 0, error_message: Optional[str] = None, result_summary: Optional[str] = None, ) -> bool: """status transition. require_owner=True 면 WHERE run_id=:rid 추가. AC-C7 (spec §4.7): claim 후 모든 status transition (success/partial/failed + exception handler + cleanup failure) 에서 owner check 필수. """ started_at = self._now() if status == "running" else None completed_at = self._now() if status in ("completed", "partial", "failed") else None now = self._now() where_clause = ( "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid" ) if require_owner: where_clause += " AND run_id = :rid" 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 = EXCLUDED.status, run_id = EXCLUDED.run_id, resolved_model = EXCLUDED.resolved_model, applicable_count = EXCLUDED.applicable_count, completed_count = EXCLUDED.completed_count, failed_count = EXCLUDED.failed_count, error_message = EXCLUDED.error_message, result_summary = EXCLUDED.result_summary, started_at = COALESCE(EXCLUDED.started_at, step_run.started_at), completed_at = COALESCE(EXCLUDED.completed_at, step_run.completed_at), updated_at = EXCLUDED.updated_at {('WHERE step_run.run_id = :rid' if require_owner else '')} 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, }) self.db.commit() row = result.fetchone() return row is not None ``` 기존 `_update_step_run` 호출 site 들을 모두 검토 — claim 후 호출 (success/partial/failed/exception/cleanup failure) 에 `require_owner=True` 명시. `run()` 메서드 (Block C 에서 재작성하지만 미리 owner-aware 도 적용): - finally / exception handler / completed / partial / failed 호출에 `require_owner=True` 추가 - [ ] **Step 3.5: owner-aware False 반환 처리 — `step.owner_lost` AppError raise (V2 patch 추가 d)** **V2 patch 추가 d (Codex IMPORTANT #4)**: `_update_step_run(require_owner=True)` 가 False 반환 시 (`row is not None` 판정 실패) caller 처리 의무 명시. False 의 의미 = 자기 owner check 실패 (다른 worker 가 steal 해갔거나, stale steal 당함). 모든 transition path 에서 surface 의무. 새 helper 추가: ```python # step_runner.py — _update_step_run 직후 def _update_step_run_strict( self, status: str, *, completed_count: int = 0, applicable_count: int = 1, failed_count: int = 0, error_message: Optional[str] = None, result_summary: Optional[str] = None, ) -> None: """owner-aware update + False 반환 시 step.owner_lost AppError raise. V2 patch 추가 d (Codex IMPORTANT #4): claim 후 transition 의 모든 path (success / partial / failed / exception handler / cleanup failure) 에서 owner check 실패 (False) 는 race lost 신호 — 자동 silent skip 금지. """ updated = self._update_step_run( status, require_owner=True, completed_count=completed_count, applicable_count=applicable_count, failed_count=failed_count, error_message=error_message, result_summary=result_summary, ) if not updated: raise AppError( code="step.owner_lost", message=( f"{self.step_id} owner check failed (run_id={self.run_id}) — " f"다른 worker 가 step_run row 를 갱신했거나 stale steal 당함. " f"transition='{status}' 미적용." ), status_code=409, ) ``` **caller 갱신 의무**: claim 후 transition 호출은 `_update_step_run_strict(...)` 사용 (require_owner=True 자동 적용 + False → AppError raise). `_update_step_run(require_owner=False)` 직접 호출은 claim 안 한 path 한정 (예: NOT_APPLICABLE 의 `_mark_not_applicable`). 추가 단위 테스트: ```python # test_step_runner_owner_aware_update.py 에 append def test_update_step_run_strict_raises_when_owner_lost(): """V2 patch 추가 d: False 반환 시 step.owner_lost AppError raise.""" from app.core.errors import AppError runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1" runner.episode_id = "e1" runner.run_id = "r-self" runner.db = MagicMock() runner._now = lambda: "2026-05-07T00:00:00+00:00" runner._resolve_model = lambda: "test-model" # fetchone 이 None — owner mismatch 시뮬 runner.db.execute.return_value.fetchone = MagicMock(return_value=None) with pytest.raises(AppError) as exc_info: runner._update_step_run_strict("completed") assert exc_info.value.code == "step.owner_lost" assert exc_info.value.status_code == 409 ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_step_runner_owner_aware_update.py -v` Expected: 3 passed (require_owner=True/False/strict-raises) - [ ] **Step 5: 회귀 — 기존 step_run 관련 테스트** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k step_run -v && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k step_runner -v` Expected: 모두 PASS - [ ] **Step 6: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_step_runner_owner_aware_update.py git commit -m "feat(step_runner): _update_step_run(require_owner) for race-safe transitions (AC-C7)" ``` --- ### Task B10: `_mark_not_applicable()` helper **Files:** - Modify: `backend/app/core/step_runner.py` - Test: `backend/tests/unit/test_resume_decision.py` (확장) **AC:** §4.8 (V2 patch I4) - [ ] **Step 1: 테스트 추가** ```python # tests/unit/test_resume_decision.py 에 append def test_mark_not_applicable_writes_db_and_checkpoint(monkeypatch): """V2 patch I4: NOT_APPLICABLE 도 DB step_run + cp 기록 (현재 동작 보존).""" from app.core.step_runner import StepRunner from unittest.mock import MagicMock runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1" runner.episode_id = "e1" runner.run_id = "r1" runner.db = MagicMock() runner._now = lambda: "2026-05-07T00:00:00+00:00" saved = [] runner.save_checkpoint = lambda data: saved.append(data) runner._update_step_run = MagicMock(return_value=True) runner._mark_not_applicable(reason="check_applicability=False") runner._update_step_run.assert_called_once() assert runner._update_step_run.call_args.args[0] == "not_applicable" assert len(saved) == 1 assert saved[0]["status"] == "not_applicable" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py::test_mark_not_applicable_writes_db_and_checkpoint -v` Expected: FAIL — `_mark_not_applicable` 미정의 - [ ] **Step 3: 메서드 구현** `backend/app/core/step_runner.py` `StepRunner` 안: ```python def _mark_not_applicable(self, reason: str = "") -> None: """NOT_APPLICABLE 처리 — claim 없이 DB step_run + cp 기록. V2 patch I4 (spec §4.8): 현재 동작 보존. claim 안 했으므로 require_owner=False. """ self._update_step_run("not_applicable", require_owner=False) self.save_checkpoint({"status": "not_applicable", "reason": reason}) ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: 모두 PASS - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_resume_decision.py git commit -m "feat(step_runner): _mark_not_applicable helper (V2 patch I4)" ``` --- ### Task B11: `_evaluate_resume_decision()` — 종합 판정자 **Files:** - Modify: `backend/app/core/step_runner.py` **AC:** AC-B1, AC-B3, AC-B4, AC-B5, AC-B7 - [ ] **Step 1: 테스트 작성 (다양한 분기)** `backend/tests/unit/test_resume_decision.py` 에 append: ```python # V2.1 patch (Codex BLOCKING #1): _get_step_run() 은 dict 반환 (V5 S7 / B0). # fixture 도 dict — tuple-style stub 사용 금지. def test_evaluate_returns_skip_when_completed_and_clean(monkeypatch): from app.core.step_runner import ResumeAction, StepRunner from unittest.mock import MagicMock runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner._get_step_run = lambda sid: { "status": "completed", "run_id": "r1", "started_at": "2026-05-07T00:00:00+00:00", "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": "2026-05-07T00:01:00+00:00", } runner.load_checkpoint = lambda: {"schema_version": 1, "config_hash": "h1", "data": {}} runner.manifest = {"schema_version": 1} runner.project_config = {} runner._check_cp_mismatch = lambda cp: None from app.core.integrity_report import CompletionReport runner._safe_verify_completion = lambda: CompletionReport( is_complete=True, missing=[], severity="clean", metadata={}, origin="clean", ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.SKIP def test_evaluate_returns_block_on_invariant_drift(): from app.core.step_runner import ResumeAction, StepRunner runner = StepRunner.__new__(StepRunner) runner.step_id = "scene_detail" runner._get_step_run = lambda sid: { "status": "completed", "run_id": "r1", "started_at": "2026-05-07T00:00:00+00:00", "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": "2026-05-07T00:01:00+00:00", } runner.load_checkpoint = lambda: {"schema_version": 7, "config_hash": "h", "data": {}} runner.manifest = {"schema_version": 7} runner.project_config = {} runner._check_cp_mismatch = lambda cp: None from app.core.integrity_report import CompletionReport runner._safe_verify_completion = lambda: CompletionReport( is_complete=False, missing=["sentinel_drifted"], severity="missing", metadata={}, origin="invariant_drift", ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.BLOCK def test_evaluate_returns_force_explicit_when_mode_force(): from app.core.step_runner import ResumeAction, StepRunner runner = StepRunner.__new__(StepRunner) runner.step_id = "test" decision = runner._evaluate_resume_decision(mode="force") assert decision.action == ResumeAction.FORCE_EXPLICIT ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: FAIL — `_evaluate_resume_decision` 미정의 - [ ] **Step 3: 종합 판정자 구현** `backend/app/core/step_runner.py` `StepRunner` 안: ```python def _evaluate_resume_decision(self, mode: str = "resume") -> "ResumeDecision": """단일 ResumeDecision 판정자. AC-B1, B7 (spec §2.3). mode='force' → FORCE_EXPLICIT (사용자 명시 force, claim 후 _execute_force). mode='resume': - status=completed + cp clean → SKIP - status=completed + verify result → origin 분기 - status=running → _evaluate_running_state (BLOCK or STALE_RUNNING_RECOVERY) - status in {failed, partial, stale, pending} → RERUN_SELF (auto-recovery) - 그 외 → RERUN_SELF (first-run) """ if mode == "force": return ResumeDecision( action=ResumeAction.FORCE_EXPLICIT, reason="user-requested force", ) existing = self._get_step_run(self.step_id) if not existing: # first-run return ResumeDecision( action=ResumeAction.RERUN_SELF, reason="first-run (no step_run row)", ) # V2.1 patch (Codex BLOCKING #1): _get_step_run() 은 V5 S7 / B0 task 에서 dict 만 반환. # tuple fallback 제거 — B0 의 grep 의무 (`existing[\d]` 0 lines) 위반 차단. status = existing.get("status") if status == "completed": cp = self.load_checkpoint() if cp is None: # cp 부재 → artifact_missing → RERUN_SELF return ResumeDecision( action=ResumeAction.RERUN_SELF, reason="completed but cp missing (artifact_missing)", origin="artifact_missing", ) # contract_drift check (schema/config_hash) mismatch = self._check_cp_mismatch(cp) if mismatch: return self._evaluate_contract_drift(mismatch) # verify_completion report = self._safe_verify_completion() if report.is_complete: return ResumeDecision( action=ResumeAction.SKIP, reason="cp clean, verify passed", origin="clean", ) # origin 별 분기 if report.origin == "contract_drift": return ResumeDecision( action=ResumeAction.BLOCK, reason=f"contract_drift: {report.missing[:3]}", origin="contract_drift", ) if report.origin == "invariant_drift": return self._evaluate_invariant_drift(report) # artifact_missing default return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=f"artifact_missing: {report.missing[:3]}", origin="artifact_missing", ) if status == "running": # V2.1 patch (Codex BLOCKING #1): existing 은 항상 dict (B0/V5 S7 contract). # tuple 변환 layer 제거. return self._evaluate_running_state(existing) if status in ("failed", "partial", "stale", "pending"): # AC-B6: auto-recovery 는 rerun_self return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=f"status={status} → auto-recovery rerun_self", origin="artifact_missing", ) # 알 수 없는 status — BLOCK (안전) return ResumeDecision( action=ResumeAction.BLOCK, reason=f"unknown status: {status!r}", ) ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: 모두 PASS - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_resume_decision.py git commit -m "feat(step_runner): _evaluate_resume_decision unified judge (AC-B1, B7)" ``` --- ### Task B12: `_execute_rerun_self()` + `_execute_force()` 분리 + `_execute_and_finalize()` 공통 (V2 patch P3 + spec V5 S2) **V2 patch P3 (Codex BLOCKING #3 + spec V5 S2)**: 이전 v1 plan 의 `_execute_rerun_self` / `_execute_force` 는 `_execute()` 결과만 return — 현 `step_runner.py:661-693` 의 finalization (final_status / exit verify / `_update_step_run` / save_checkpoint / `_reset_recovery_counter` / `modifies_checkpoints` cascade) 6 단계 모두 손실. spec V5 S2 의 공통 `_execute_and_finalize()` helper 패턴 도입 — `_execute_rerun_self` / `_execute_force` 둘 다 finalize 공유. **P3 보강 체크리스트 (구현자 의무)** — 현 run() try 블록 (line 652-717) 의 다음 6 단계가 `_execute_and_finalize` 에 모두 1:1 이식: - [ ] (1) `result = self._execute(mode)` + `self._last_execute_result = result` - [ ] (2) `final_status` 계산 (failed/partial/completed) — `step_runner.py:661-665` 로직 - [ ] (3) exit verify (final_status='completed' 시 `_safe_verify_completion` → 실패 시 partial 격상) — `step_runner.py:670-677` - [ ] (4) `_update_step_run_strict(final_status, ...)` (require_owner=True 자동, AC-C7 + 추가 d) — `step_runner.py:679-688` - [ ] (5) `save_checkpoint({"status": final_status, **result})` — `step_runner.py:689` - [ ] (6) `_reset_recovery_counter()` (final_status=='completed' 시) — `step_runner.py:692-693` - [ ] (7) `modifies_checkpoints` cascade policy (1-pass default) — `step_runner.py:703-713` - [ ] (8) `return {"status": final_status, "result": result}` **Files:** - Modify: `backend/app/core/step_runner.py` (`_execute_rerun_self` / `_execute_force` / `_execute_and_finalize` 추가) **AC:** AC-B6 / AC-B8 / AC-C7 - [ ] **Step 1: 테스트 추가** `backend/tests/unit/test_resume_decision.py` 에 append: ```python def test_execute_rerun_self_does_not_invalidate_downstream(): """AC-B6: rerun_self 는 cleanup_artifacts + invalidate_downstream 호출 안 함.""" from app.core.step_runner import StepRunner from unittest.mock import MagicMock runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.run_id = "r1" runner.cleanup_artifacts = MagicMock() runner.invalidate_downstream = MagicMock() runner.clear_checkpoint = MagicMock() # _execute_and_finalize 를 stub 해 단위 격리 — finalization 검증은 별도 테스트 runner._execute_and_finalize = MagicMock(return_value={"status": "completed"}) runner.build_opik_metadata = lambda: {} runner._execute_rerun_self() runner.cleanup_artifacts.assert_not_called() runner.invalidate_downstream.assert_not_called() runner.clear_checkpoint.assert_not_called() runner._execute_and_finalize.assert_called_once() def test_execute_force_calls_cleanup_then_finalize(): """AC-B6: force 는 cleanup_artifacts + invalidate_downstream(delete_cp=True) + finalize.""" from app.core.step_runner import StepRunner from unittest.mock import MagicMock runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.run_id = "r1" cleanup_report = MagicMock() cleanup_report.deleted_db_rows = 0 cleanup_report.deleted_files = 0 cleanup_report.targets = [] runner.cleanup_artifacts = MagicMock(return_value=cleanup_report) runner.invalidate_downstream = MagicMock() runner.clear_checkpoint = MagicMock() runner._execute_and_finalize = MagicMock(return_value={"status": "completed"}) runner.build_opik_metadata = lambda: {} runner._update_step_run_strict = MagicMock() runner._execute_force() runner.cleanup_artifacts.assert_called_once() runner.invalidate_downstream.assert_called_once() runner._execute_and_finalize.assert_called_once() def test_execute_and_finalize_persists_all_six_steps(): """V5 S2 + P3 보강: _execute_and_finalize 가 6 단계 모두 호출. final_status / exit verify / _update_step_run_strict / save_checkpoint / _reset_recovery_counter / modifies_checkpoints cascade. """ from app.core.step_runner import StepRunner from app.core.integrity_report import CompletionReport from unittest.mock import MagicMock runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.run_id = "r1" runner._execute = MagicMock(return_value={ "completed_count": 1, "applicable_count": 1, "failed_count": 0, }) runner._safe_verify_completion = MagicMock(return_value=CompletionReport( is_complete=True, missing=[], severity="clean", metadata={}, origin="clean", )) runner._update_step_run_strict = MagicMock() runner.save_checkpoint = MagicMock() runner._reset_recovery_counter = MagicMock() runner.invalidate_downstream = MagicMock() runner.manifest = {"modifies_checkpoints": [], "invalidate_downstream_on_edit": False} result = runner._execute_and_finalize() runner._execute.assert_called_once() runner._safe_verify_completion.assert_called_once() # final_status='completed' 시 runner._update_step_run_strict.assert_called_once() # require_owner=True 자동 runner.save_checkpoint.assert_called_once() runner._reset_recovery_counter.assert_called_once() # completed 시 reset assert result["status"] == "completed" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: FAIL — `_execute_rerun_self` / `_execute_force` / `_execute_and_finalize` 미정의 - [ ] **Step 3: 메서드 추가 — `_execute_and_finalize()` 공통 + `_execute_rerun_self()` + `_execute_force()`** `backend/app/core/step_runner.py` `StepRunner` 안에 추가: ```python def _execute_rerun_self(self) -> Dict[str, Any]: """auto-recovery 경로 — 자기 step 만 rerun. AC-B6 (spec §4.2). force 와 다른 점 (사전 정책): - cleanup_artifacts() 호출 안 함 - invalidate_downstream() 호출 안 함 - downstream cp 보존 - force-cleared marker 안 박힘 finalize 는 _execute_and_finalize() 공통 helper 에서 (V5 S2). """ return self._execute_and_finalize(execute_mode="resume") def _execute_force(self) -> Dict[str, Any]: """사용자 명시 force — 기존 cleanup + invalidate_downstream cascade 경로. AC-B6 의 대척 — force 는 의도적으로 downstream invalidate. finalize 는 _execute_and_finalize() 공통 helper 에서 (V5 S2). """ # cleanup_artifacts (override 한 step 만 실제 동작) try: cleanup_report = self.cleanup_artifacts() except Exception as exc: # cleanup 실패 — claim 한 status 를 failed 로 (require_owner=True) self._update_step_run_strict( "failed", error_message=f"cleanup_artifacts crashed: {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, ) # downstream invalidate + cp clear self.invalidate_downstream(delete_checkpoints=True) self.clear_checkpoint() return self._execute_and_finalize(execute_mode="force") def _execute_and_finalize(self, *, execute_mode: str = "resume") -> Dict[str, Any]: """V5 S2 + P3 보강: rerun_self / force 두 경로 공통 finalize. 이식한 6 단계 (현 step_runner.py:652-717): (1) _execute() 호출 + _last_execute_result attribute set (2) final_status 계산 (failed/partial/completed) (3) exit verify (final_status='completed' 시) → 실패 시 partial 격상 (AC-B8) (4) _update_step_run_strict(final_status, ...) — require_owner=True (AC-C7) (5) save_checkpoint({"status": final_status, **result}) (6) _reset_recovery_counter() (final_status='completed' 시) (7) modifies_checkpoints cascade policy (1-pass default) """ from app.modules.llm.llm_client import set_opik_context set_opik_context(self.build_opik_metadata()) # V2.1.1 patch (Codex NEEDS_SMALL_PATCH BLOCKING): inner exit verify catch 가 specific # error_message 로 status='failed' 기록한 경우 outer except 가 같은 owner 로 generic # message 로 덮는 race 차단 — local flag 로 한 번만 기록 보장. failed_recorded = False try: # (1) result = self._execute(mode=execute_mode) self._last_execute_result = result # (2) completed = result.get("completed_count", 1) total = result.get("applicable_count", 1) failed = result.get("failed_count", 0) final_status = "completed" if failed == 0 else ("partial" if completed > 0 else "failed") # (3) exit verify (AC-B8) # V2.1 patch (Codex BLOCKING #2 + spec V5 S6): _safe_verify_completion 이 # `step.verify_crashed` AppError raise 시 status='failed' 기록 후 re-raise. # 그렇지 않으면 finalize 가 transition 안 하고 run() 까지 올라가 status 가 # 'running' 으로 stale leak. 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, verify_exc.code, verify_exc.message, ) try: self._update_step_run_strict( "failed", completed_count=completed, applicable_count=total, failed_count=failed, error_message=( f"exit verify crashed: {verify_exc.code} — {verify_exc.message}" )[:2000], ) failed_recorded = True # V2.1.1 patch: outer except 의 generic overwrite 차단 except AppError as owner_exc: logger.warning( "exit verify failed transition owner_lost: %s", owner_exc, ) 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" # (4) status transition — owner-aware strict (AC-C7 + V2 추가 d) 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], ) # (5) save_checkpoint self.save_checkpoint({"status": final_status, **result}) # (6) recovery counter reset (정상 완료 시) if final_status == "completed": self._reset_recovery_counter() # (7) modifies_checkpoints cascade 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 AppError: # V2.1 patch (Codex BLOCKING #2): AppError 가 (1) _execute / (4)~(7) finalize path # 에서 raise 시 status='failed' 기록. (3) exit verify 는 자체 specific message 로 # 이미 기록 — V2.1.1 patch 로 failed_recorded flag 로 중복 방지. if not failed_recorded: try: self._update_step_run_strict( "failed", error_message="AppError in _execute_and_finalize", ) except AppError: pass # owner_lost — 이미 다른 path 에서 기록됨 raise except Exception as exc: # V2.1 patch (Codex BLOCKING #2): _execute() 또는 finalize 단계의 unexpected # exception (KeyError / SQLAlchemyError 등) — status='failed' 기록 후 re-raise. # 이 처리 누락 시 status='running' 으로 leak 되어 다음 resume 에서 stale 분기. # V2.1.1 patch: failed_recorded 시 specific exit verify message 보존. import traceback if not failed_recorded: try: self._update_step_run_strict( "failed", error_message=str(exc)[:2000], ) except AppError as owner_exc: logger.warning( "exception path failed transition owner_lost: %s (original=%s)", owner_exc, 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) ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_resume_decision.py -v` Expected: 모두 PASS - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_resume_decision.py git commit -m "feat(step_runner): split rerun_self/force + _execute_and_finalize common (AC-B6, B8, C7 / V5 S2)" ``` --- ### Task B13: `step_execution_service.py` 자체 skip 판단 제거 **Files:** - Modify: `backend/app/services/step_execution_service.py:169-200` **AC:** AC-B1, AC-B7 - [ ] **Step 1: 변경 전 코드 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k step_execution_service -v` 기존 통과 테스트 목록 capture (회귀 비교 baseline). - [ ] **Step 2: `start_step()` 의 자체 skip 블록 제거** `backend/app/services/step_execution_service.py:169-200` (mode == "resume" + completed + cp 검증 + skip return) 부분을 다음으로 교체: ```python # B 1: 자체 resume 판단 제거 (AC-B1). # StepRunner.run() 의 _evaluate_resume_decision 단일 판정자에 위임. # 모든 resume 결정은 background_worker → StepRunner.run 단일 경로. # (기존 169-200 블록 모두 삭제 — disabled 차단 + gate 만 남김) # (170 line) gate 만 남기고 곧장 background_job submit 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} ``` - [ ] **Step 3: 두 path 회귀 테스트 (V2 patch P6 + spec V5 S3)** **V2 patch P6 (Codex IMPORTANT #2 + spec V5 S3)**: 동일 runner 의 `_evaluate_resume_decision()` 두 번 호출은 deterministic 검증만 — `start_step()` (background_worker → `StepRunner.run`) 와 `run_steps_batch()` 의 **두 path 차이를 검증하지 못함**. spec §9 AC-B7 (V5 S3) 에 따라 두 path 각각 stub 으로 진입해 같은 ResumeDecision/ResumeAction 도달 + 같은 claim 동작 (또는 SKIP 조기 return) 검증. ```python # backend/tests/integration/test_resume_single_vs_run_all_consistency.py """single-step (start_step) vs run-all (run_steps_batch) 동일성 — AC-B7 (V5 S3).""" import pytest from unittest.mock import MagicMock, patch @pytest.fixture def captured_runs(monkeypatch): """StepRunner.run 호출 시점의 (step_id, mode) 와 internal decision 을 capture. 두 path 모두 같은 StepRunner.run 단일 진입점에 도달함을 검증 (V5 S3). """ records = [] def stub_run(self, mode="resume"): # 실제 _evaluate_resume_decision 호출 (StepRunner instance method) decision = self._evaluate_resume_decision(mode) records.append({ "step_id": self.step_id, "mode": mode, "decision_action": decision.action.value, "decision_origin": decision.origin, }) return {"status": "skipped" if decision.action.value == "skip" else "completed"} monkeypatch.setattr("app.core.step_runner.StepRunner.run", stub_run) return records @pytest.fixture def make_runner_completed_clean(): """completed + cp clean 상태의 StepRunner 인스턴스 빌더 (생성자 우회).""" def _build(step_id: str = "scene_detail"): from app.core.step_runner import StepRunner from app.core.integrity_report import CompletionReport runner = StepRunner.__new__(StepRunner) runner.step_id = step_id runner.project_id = "p" runner.episode_id = "e" runner.run_id = "r-test" # V2 patch P4: dict shape (B0 완료 후) runner._get_step_run = lambda sid: { "status": "completed", "run_id": "r-prev", "started_at": "2026-05-08T00:00:00+00:00", "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": "2026-05-08T00:01:00+00:00", } runner.load_checkpoint = lambda: { "schema_version": 1, "config_hash": "h", "data": {} } runner.manifest = {"schema_version": 1} runner.project_config = {} runner._check_cp_mismatch = lambda cp: None runner._safe_verify_completion = lambda: CompletionReport( is_complete=True, missing=[], severity="clean", metadata={}, origin="clean", ) runner.check_gate = lambda: None runner.check_applicability = lambda: True return runner return _build def test_start_step_path_makes_skip_decision(captured_runs, make_runner_completed_clean, monkeypatch): """V5 S3 path 1: step_execution_service.start_step → _background_worker → StepRunner.run → SKIP. start_step 의 background_worker submit + 실행을 inline (직접 호출) 로 시뮬. """ runner = make_runner_completed_clean() # path 1: background_worker 경로 simulate — StepRunner.run("resume") 직접 진입 result = runner.run(mode="resume") assert result["status"] == "skipped" assert any(r["step_id"] == "scene_detail" and r["mode"] == "resume" for r in captured_runs) rec = next(r for r in captured_runs if r["step_id"] == "scene_detail") assert rec["decision_action"] == "skip" assert rec["decision_origin"] == "clean" def test_run_steps_batch_path_makes_skip_decision(captured_runs, make_runner_completed_clean): """V5 S3 path 2: run_steps_batch → 각 step 의 StepRunner.run → SKIP. run_steps_batch 가 같은 StepRunner.run("resume") 진입점을 사용함을 검증. """ # path 2: run_steps_batch 의 inner loop 시뮬 — 다수 runner 의 run 호출 runners = [make_runner_completed_clean(sid) for sid in ("scene_detail", "shot_dependency_t2i")] for r in runners: r.run(mode="resume") # 두 step 모두 SKIP 도달 actions = [r["decision_action"] for r in captured_runs] assert all(a == "skip" for a in actions) assert len(actions) == 2 def test_two_paths_make_identical_decision(captured_runs, make_runner_completed_clean): """V5 S3: 두 path 가 같은 (step_id, mode, sentinel state) 에서 같은 ResumeAction 도달.""" # path 1 — start_step 시뮬 r1 = make_runner_completed_clean() r1.run(mode="resume") # path 2 — run_steps_batch 시뮬 (같은 설정) r2 = make_runner_completed_clean() r2.run(mode="resume") actions = [r["decision_action"] for r in captured_runs] origins = [r["decision_origin"] for r in captured_runs] assert len(set(actions)) == 1, f"두 path 의 ResumeAction 불일치: {actions}" assert len(set(origins)) == 1, f"두 path 의 origin 불일치: {origins}" ``` **검증 의무**: 두 fixture (`test_start_step_path_*` + `test_run_steps_batch_path_*`) 가 같은 `_evaluate_resume_decision` 진입점에 도달함을 포착. start_step path 의 background_worker submit + run_steps_batch path 의 sequential 호출은 inline (직접 `runner.run`) 으로 시뮬 — 실제 background_worker 의 thread/process 분리는 별도 통합 테스트 (Task C3) 로. - [ ] **Step 4: 테스트 실행** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_resume_single_vs_run_all_consistency.py tests/unit/ -k step_execution_service -v` Expected: 신규 3건 통과 + 기존 회귀 0 - [ ] **Step 5: commit** ```bash git add backend/app/services/step_execution_service.py backend/tests/integration/test_resume_single_vs_run_all_consistency.py git commit -m "refactor(step_execution_service): remove self-skip judgment, defer to StepRunner (AC-B1, B7 / V5 S3)" ``` --- ### Task B14: Block B 통합 회귀 — 광범위 테스트 **Files:** - (테스트 추가 없음 — 기존 + 신규 통합 실행) **AC:** AC-B1~B9 종합 - [ ] **Step 1: Block B 단위 테스트 전체** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_completion_report_origin.py tests/unit/test_resume_decision.py tests/unit/test_running_state_evaluation.py tests/unit/test_legacy_schema_bump_allowlist.py tests/unit/test_step_runner_owner_aware_update.py tests/unit/test_step_runner_verify_crashed.py tests/unit/test_scene_detail_verify_origin.py tests/unit/test_config_running_timeout.py -v` Expected: 모두 PASS - [ ] **Step 2: 기존 step_runner / scene_detail 회귀** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -k 'step_runner or scene_detail or step_run' -v` Expected: 회귀 0 — 기존 통과 테스트 모두 PASS - [ ] **Step 3: Block B integration** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_resume_single_vs_run_all_consistency.py -v` Expected: PASS - [ ] **Step 4: Block A + B 합산 smoke** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -v --tb=short 2>&1 | tail -30` Expected: 전체 회귀 0 — 통과 카운트 기존 + Block A/B 신규 - [ ] **Step 5: commit (회귀 OK 확인 marker)** ```bash git commit --allow-empty -m "test: Block A+B unit/integration regression confirmed (AC-B1..B9 all green)" ``` --- ## Block C — Concurrency lock (~0.5d) ### Task C1: `_try_claim_running()` atomic INSERT ON CONFLICT **Files:** - Modify: `backend/app/core/step_runner.py` - Test: `backend/tests/unit/test_try_claim_running.py` (신규) **AC:** AC-C1, AC-C6, AC-C9, AC-C10, AC-C11 - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_try_claim_running.py """_try_claim_running atomic INSERT ON CONFLICT — AC-C1, C6, C9, C10, C11.""" from unittest.mock import MagicMock from app.core.step_runner import StepRunner def test_first_run_inserts_new_row(): """AC-C6: row 없을 때 INSERT 경로로 claim 성공.""" runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1" runner.episode_id = "e1" runner.run_id = "r1" runner.db = MagicMock() # mock fetchone — INSERT 성공 row 반환 result_mock = MagicMock() result_mock.fetchone.return_value = ("r1",) runner.db.execute.return_value = result_mock ok = runner._try_claim_running() assert ok is True def test_existing_running_returns_false_when_no_steal(): """이미 running 이고 stale 아닐 때 → claim 실패.""" runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1"; runner.episode_id = "e1"; runner.run_id = "r1" runner.db = MagicMock() result_mock = MagicMock() result_mock.fetchone.return_value = None # WHERE 조건 미충족 → row 0 runner.db.execute.return_value = result_mock ok = runner._try_claim_running() assert ok is False def test_stale_steal_with_expected_match(): """AC-C8: STALE_RUNNING_RECOVERY 시 expected_started_at + expected_run_id 매칭.""" runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1"; runner.episode_id = "e1"; runner.run_id = "r-new" runner.db = MagicMock() result_mock = MagicMock() result_mock.fetchone.return_value = ("r-new",) runner.db.execute.return_value = result_mock ok = runner._try_claim_running( allow_stale_steal=True, expected_started_at="2026-05-07T10:00:00+00:00", expected_run_id="r-old", ) assert ok is True # SQL 의 bind params 검증 call_args = runner.db.execute.call_args bind_params = call_args[0][1] assert bind_params["allow_stale_steal"] is True assert bind_params["expected_started_at"] == "2026-05-07T10:00:00+00:00" assert bind_params["expected_run_id"] == "r-old" def test_no_sql_cast_in_query(): """AC-C10: claim SQL 에 ::timestamptz cast 0건.""" runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1"; runner.episode_id = "e1"; runner.run_id = "r1" runner.db = MagicMock() result_mock = MagicMock() result_mock.fetchone.return_value = None runner.db.execute.return_value = result_mock runner._try_claim_running() sql_text = str(runner.db.execute.call_args[0][0]) assert "::timestamptz" not in sql_text, f"SQL contains cast: {sql_text}" assert "::interval" not in sql_text def test_uses_fetchone_not_rowcount(): """AC-C11: result.fetchone() is not None 으로 판정.""" runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1"; runner.episode_id = "e1"; runner.run_id = "r1" runner.db = MagicMock() result_mock = MagicMock() result_mock.fetchone.return_value = ("r1",) runner.db.execute.return_value = result_mock ok = runner._try_claim_running() assert ok is True # fetchone 호출 검증 result_mock.fetchone.assert_called_once() def test_insert_columns_includes_all_not_null(): """AC-C9: INSERT 컬럼 목록에 id, project_id, episode_id, step_id, run_id, status, started_at, created_at, updated_at 모두 명시.""" runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1"; runner.episode_id = "e1"; runner.run_id = "r1" runner.db = MagicMock() result_mock = MagicMock() result_mock.fetchone.return_value = None runner.db.execute.return_value = result_mock runner._try_claim_running() sql_text = str(runner.db.execute.call_args[0][0]) for col in ("id", "project_id", "episode_id", "step_id", "run_id", "status", "started_at", "created_at", "updated_at"): assert col in sql_text, f"INSERT 에 {col!r} 누락" ``` - [ ] **Step 2: 테스트 실패 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_try_claim_running.py -v` Expected: FAIL — `_try_claim_running` 미정의 - [ ] **Step 3: 메서드 구현** `backend/app/core/step_runner.py` `StepRunner` 안: ```python 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. AC-C1, C6, C8, C9, C10, C11 (spec §5.2): - first-run (row 없음) → INSERT 경로 - 기존 row + status != 'running' → DO UPDATE 경로 - STALE_RUNNING_RECOVERY (allow_stale_steal=True) → expected-match steal SQL cast 0 (V3 patch B1) — timeout 판정은 Python 에서 완료. fetchone() 으로 판정 (V3 patch I2) — rowcount driver 차이 회피. Returns: True (claim 성공) / False (다른 worker 이미 잡음 또는 expected mismatch). """ import uuid new_step_run_id = str(uuid.uuid4()) sql = text(""" INSERT INTO step_run ( id, project_id, episode_id, step_id, run_id, status, started_at, created_at, updated_at ) VALUES ( :new_step_run_id, :pid, :eid, :sid, :run_id, 'running', NOW()::text, NOW()::text, NOW()::text ) ON CONFLICT (project_id, episode_id, step_id) DO UPDATE SET run_id = EXCLUDED.run_id, status = 'running', started_at = EXCLUDED.started_at, updated_at = EXCLUDED.updated_at, recovery_count = step_run.recovery_count + CASE WHEN step_run.status = 'running' THEN 1 ELSE 0 END WHERE step_run.status != 'running' OR ( :allow_stale_steal IS TRUE AND step_run.status = 'running' AND step_run.started_at = :expected_started_at AND step_run.run_id = :expected_run_id ) RETURNING run_id """) result = self.db.execute(sql, { "new_step_run_id": new_step_run_id, "pid": self.project_id, "eid": self.episode_id, "sid": self.step_id, "run_id": self.run_id, "allow_stale_steal": allow_stale_steal, "expected_started_at": expected_started_at, "expected_run_id": expected_run_id, }) self.db.commit() row = result.fetchone() return row is not None ``` - [ ] **Step 4: 테스트 통과 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_try_claim_running.py -v` Expected: 6 passed - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/unit/test_try_claim_running.py git commit -m "feat(step_runner): _try_claim_running with INSERT ON CONFLICT + expected-match steal (AC-C1, C6-C11)" ``` --- ### Task C2: `run()` 흐름 재작성 — claim 시점 + decision 분기 + finalize 위임 (V2 patch P3 + 추가 a) **V2 patch P3 (Codex BLOCKING #3 + spec V5 S2)**: run() 의 try 블록은 `_execute_rerun_self` / `_execute_force` 가 내부의 `_execute_and_finalize()` 로 finalize 까지 포함하므로, run() 은 단순 분기 + exception 처리만. **V2 patch 추가 a (Codex BLOCKING #5 + spec V5 S6)**: AppError 의 `extra=...` keyword 제거 — `errors.py:5` signature `(code, message, status_code)` 만. origin 정보는 message 에 inline. **Files:** - Modify: `backend/app/core/step_runner.py` (`run()` 메서드 재작성) **AC:** AC-B1, AC-B6, AC-C1, AC-C2, AC-C7 - [ ] **Step 1: `run()` 본문 변경 — 새 흐름** `backend/app/core/step_runner.py` 의 `run()` 메서드 (line ~518) 를 다음으로 교체: ```python def run(self, mode: str = "resume") -> Dict[str, Any]: """단계 실행 — 신규 4-phase 흐름 (V3 patch B1 + V5 S2): 1) gate / applicability (non-mutating) 2) ResumeDecision 평가 (non-mutating) 3) atomic claim (RERUN_SELF / FORCE_EXPLICIT / STALE_RUNNING_RECOVERY 만) 4) _execute_rerun_self() / _execute_force() 분기 — 두 메서드 모두 내부에서 _execute_and_finalize() 호출 (final_status / exit verify / _update_step_run_strict / save_checkpoint / _reset_recovery_counter / mods cascade — 6 단계 보존). AC-B1, B6, B7, C1, C2, C7. """ # 1. gate / applicability (non-mutating) self.check_gate() if not self.check_applicability(): # V2 patch I4: NOT_APPLICABLE 도 DB+cp 기록 self._mark_not_applicable(reason="check_applicability=False") return {"status": "not_applicable"} # 2. ResumeDecision 평가 (non-mutating) decision = self._evaluate_resume_decision(mode) if decision.action == ResumeAction.SKIP: return {"status": "skipped", "reason": decision.reason} if decision.action == ResumeAction.BLOCK: # V2 patch 추가 a (V5 S6): AppError signature 만 — extra keyword 금지. # origin 정보는 message 에 inline. from app.core.errors import AppError raise AppError( code="step.resume_blocked", message=f"{decision.reason} (origin={decision.origin})", status_code=409, ) if decision.action == ResumeAction.NOT_APPLICABLE: self._mark_not_applicable(reason=decision.reason) return {"status": "not_applicable", "reason": decision.reason} # 3. atomic claim (RERUN_SELF / FORCE_EXPLICIT / STALE_RUNNING_RECOVERY) is_stale_steal = (decision.action == ResumeAction.STALE_RUNNING_RECOVERY) if not 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, ): from app.core.errors import AppError raise AppError( code="step.already_running", message=f"{self.step_id} 이 이미 다른 worker 에서 실행 중 (또는 stale read 후 갱신됨)", status_code=409, ) logger.info( "Step %s started (run_id=%s, model=%s, decision=%s)", self.step_id, self.run_id, self._resolve_model(), decision.action.value, ) # 4. 실행 분기 — _execute_* 가 내부에서 _execute_and_finalize 호출 (V5 S2) try: if decision.action == ResumeAction.FORCE_EXPLICIT: return self._execute_force() else: # RERUN_SELF / STALE_RUNNING_RECOVERY return self._execute_rerun_self() except AppError: # owner_lost / verify_crashed / 기존 AppError 는 그대로 전달 — _execute_and_finalize # 내부 _update_step_run_strict 가 transition 처리. 추가 status update 안 함. raise except Exception as exc: # AC-C7 + V2 추가 d: 예측 못 한 exception 시 owner-aware failed transition. # _execute_and_finalize 가 도달 못 한 경우 (e.g. cleanup_artifacts crash 후 재진입) 한정. import traceback try: self._update_step_run_strict( "failed", error_message=str(exc), ) except AppError as owner_exc: # owner_lost — 다른 worker 가 이미 transition 함. 조용히 surface. logger.warning( "Step %s exception path owner_lost: %s (original=%s)", self.step_id, owner_exc, exc, ) logger.error( "Step %s failed: %s\n%s", self.step_id, exc, traceback.format_exc(), ) raise ``` - [ ] **Step 1.5: AppError extra grep 검증 (V2 patch 추가 a)** ```bash # spec V5 S6 의 grep 의무 — 0 lines 검증 rg "AppError\(.*extra=" backend/app/ backend/tests/ docs/superpowers/ # Expected: 0 lines ``` - [ ] **Step 2: 통합 테스트 작성** ```python # backend/tests/integration/test_step_runner_run_flow.py """run() 새 흐름 통합 — AC-B1, C1, C2, C7.""" import pytest from unittest.mock import MagicMock, patch from app.core.errors import AppError from app.core.step_runner import StepRunner, ResumeAction, ResumeDecision def _runner_with_decision(action: ResumeAction, **kw): runner = StepRunner.__new__(StepRunner) runner.step_id = "test" runner.project_id = "p1" runner.episode_id = "e1" runner.run_id = "r1" runner.db = MagicMock() runner._resolve_model = lambda: "gpt" runner.check_gate = lambda: None runner.check_applicability = lambda: True runner._evaluate_resume_decision = lambda mode: ResumeDecision(action=action, reason="t", **kw) runner._try_claim_running = MagicMock(return_value=True) runner._execute_rerun_self = MagicMock(return_value={"status": "completed"}) runner._execute_force = MagicMock(return_value={"status": "completed"}) runner._mark_not_applicable = MagicMock() runner._update_step_run = MagicMock() return runner def test_skip_does_not_claim(): runner = _runner_with_decision(ResumeAction.SKIP) result = runner.run("resume") assert result["status"] == "skipped" runner._try_claim_running.assert_not_called() def test_block_raises_409_without_claim(): runner = _runner_with_decision(ResumeAction.BLOCK) with pytest.raises(AppError) as exc: runner.run("resume") assert exc.value.code == "step.resume_blocked" runner._try_claim_running.assert_not_called() def test_rerun_self_claims_then_executes(): runner = _runner_with_decision(ResumeAction.RERUN_SELF) result = runner.run("resume") runner._try_claim_running.assert_called_once_with( allow_stale_steal=False, expected_started_at=None, expected_run_id=None, ) runner._execute_rerun_self.assert_called_once() runner._execute_force.assert_not_called() def test_force_explicit_calls_execute_force(): runner = _runner_with_decision(ResumeAction.FORCE_EXPLICIT) result = runner.run("force") runner._execute_force.assert_called_once() runner._execute_rerun_self.assert_not_called() def test_stale_running_recovery_passes_expected_fields(): runner = _runner_with_decision( ResumeAction.STALE_RUNNING_RECOVERY, expected_started_at="2026-05-07T10:00:00+00:00", expected_run_id="r-old", ) runner.run("resume") runner._try_claim_running.assert_called_once_with( allow_stale_steal=True, expected_started_at="2026-05-07T10:00:00+00:00", expected_run_id="r-old", ) runner._execute_rerun_self.assert_called_once() def test_claim_failure_raises_already_running(): runner = _runner_with_decision(ResumeAction.RERUN_SELF) runner._try_claim_running.return_value = False with pytest.raises(AppError) as exc: runner.run("resume") assert exc.value.code == "step.already_running" def test_exception_in_execute_uses_owner_aware_failed_update(): runner = _runner_with_decision(ResumeAction.RERUN_SELF) runner._execute_rerun_self.side_effect = RuntimeError("boom") with pytest.raises(RuntimeError): runner.run("resume") # AC-C7: exception path 도 require_owner=True runner._update_step_run.assert_called() call_kwargs = runner._update_step_run.call_args.kwargs assert call_kwargs.get("require_owner") is True ``` - [ ] **Step 3: 테스트 실행** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_step_runner_run_flow.py -v` Expected: 7 passed - [ ] **Step 4: 광범위 회귀 — 기존 step_runner.run 호출자** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/ -k 'step_runner or run_steps_batch or step_execution' -v` Expected: 회귀 0 - [ ] **Step 5: commit** ```bash git add backend/app/core/step_runner.py backend/tests/integration/test_step_runner_run_flow.py git commit -m "feat(step_runner): rewrite run() flow with claim + decision dispatch (AC-B1, C1-C2, C7)" ``` --- ### Task C3: 동시 worker 회귀 테스트 **Files:** - Create: `backend/tests/integration/test_concurrent_claim.py` **AC:** AC-C5 - [ ] **Step 1: 통합 테스트 작성** ```python # backend/tests/integration/test_concurrent_claim.py """동시 worker claim — LLM/image 호출 1회만 (AC-C5).""" import threading import time import uuid import pytest from sqlalchemy import text from app.core.database import SessionLocal @pytest.fixture def step_run_seed(): """test step_run row 미리 정리.""" pid = "test-concurrent-pid-" + str(uuid.uuid4())[:8] eid = "test-concurrent-eid-" + str(uuid.uuid4())[:8] sid = "concurrent_test_step" yield pid, eid, sid # cleanup with SessionLocal() as db: db.execute(text( "DELETE FROM step_run WHERE project_id=:pid AND episode_id=:eid AND step_id=:sid" ), {"pid": pid, "eid": eid, "sid": sid}) db.commit() def test_concurrent_claim_only_one_succeeds(step_run_seed): """두 thread 가 동시에 claim 시도 → 한 쪽만 성공.""" from app.core.step_runner import StepRunner pid, eid, sid = step_run_seed results = [] barrier = threading.Barrier(2) def worker(run_id): with SessionLocal() as db: runner = StepRunner.__new__(StepRunner) runner.step_id = sid runner.project_id = pid runner.episode_id = eid runner.run_id = run_id runner.db = db barrier.wait() ok = runner._try_claim_running() results.append((run_id, ok)) t1 = threading.Thread(target=worker, args=("worker-1",)) t2 = threading.Thread(target=worker, args=("worker-2",)) t1.start(); t2.start() t1.join(); t2.join() successes = [r for r in results if r[1] is True] failures = [r for r in results if r[1] is False] assert len(successes) == 1, f"동시 claim 한 쪽만 성공해야 함: {results}" assert len(failures) == 1 ``` - [ ] **Step 2: 테스트 실행** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_concurrent_claim.py -v` Expected: PASS — 한 쪽만 success (주의: production PG 에서 동작. test PG (`theroad_test`) 가 별도면 fixture 가 자동으로 cleanup) - [ ] **Step 3: commit** ```bash git add backend/tests/integration/test_concurrent_claim.py git commit -m "test(step_runner): concurrent claim regression — only one succeeds (AC-C5)" ``` --- ### Task C4: audit_stale_running.py — 배포 전 audit script **Files:** - Create: `backend/scripts/audit_stale_running.py` **AC:** §7.3 (V3 patch B1 SQL cast 제거) - [ ] **Step 1: script 작성** ```python #!/usr/bin/env python """배포 전 audit — running status row 분석 + parse 실패 detection. V3 patch B1 (spec §7.3): SQL cast 0 — application 레벨 datetime parse. invalid text started_at row 도 안전하게 회수. Usage: cd backend && PYTHONPATH=. .venv/bin/python scripts/audit_stale_running.py cd backend && STEP_RUNNING_TIMEOUT_SECONDS=7200 PYTHONPATH=. .venv/bin/python scripts/audit_stale_running.py """ import os import sys from datetime import datetime, timezone import sqlalchemy as sa TIMEOUT_SECONDS = int(os.environ.get("STEP_RUNNING_TIMEOUT_SECONDS", "3600")) def main(): from app.core.database import SessionLocal with SessionLocal() as db: rows = db.execute(sa.text(""" SELECT id, project_id, episode_id, step_id, run_id, started_at, updated_at, recovery_count FROM step_run WHERE status = 'running' ORDER BY started_at ASC NULLS FIRST """)).fetchall() now = datetime.now(timezone.utc) healthy, stale, parse_failed, null_started = [], [], [], [] for r in rows: if r.started_at is None: null_started.append(r) continue try: st = datetime.fromisoformat(r.started_at.replace("Z", "+00:00")) elapsed = (now - st).total_seconds() if elapsed > TIMEOUT_SECONDS: stale.append((r, elapsed)) else: healthy.append((r, elapsed)) except (ValueError, TypeError) as exc: parse_failed.append((r, str(exc))) print(f"=== Running step_run audit (timeout={TIMEOUT_SECONDS}s) ===\n") print(f"[HEALTHY] {len(healthy)} rows (elapsed < {TIMEOUT_SECONDS}s)") for r, elapsed in healthy: print(f" step={r.step_id} run_id={r.run_id} elapsed={elapsed:.0f}s pid={r.project_id} eid={r.episode_id}") print(f"\n[STALE] {len(stale)} rows (elapsed > {TIMEOUT_SECONDS}s — STALE_RUNNING_RECOVERY 대상)") for r, elapsed in stale: print(f" step={r.step_id} run_id={r.run_id} elapsed={elapsed:.0f}s started_at={r.started_at}") print(f" 수동 reset SQL:") print(f" UPDATE step_run SET status='failed', error_message='pre-B deploy stale running reset', updated_at=NOW()::text WHERE id='{r.id}';") print(f"\n[PARSE_FAILED] {len(parse_failed)} rows (수동 investigation 필수)") for r, err in parse_failed: print(f" step={r.step_id} started_at={r.started_at!r} err={err}") print(f"\n[NULL_STARTED] {len(null_started)} rows (수동 investigation 필수)") for r in null_started: print(f" step={r.step_id} run_id={r.run_id} updated_at={r.updated_at}") # exit code: 0 (clean) / 1 (stale 또는 parse_failed 또는 null_started 존재) if stale or parse_failed or null_started: return 1 return 0 if __name__ == "__main__": sys.exit(main()) ``` - [ ] **Step 2: script 실행 (현재 DB 기준 smoke)** ```bash cd backend && PYTHONPATH=. .venv/bin/python scripts/audit_stale_running.py ``` Expected: 출력 — healthy/stale/parse_failed/null_started 카테고리별 row 수 + step_id 목록. - [ ] **Step 3: commit** ```bash git add backend/scripts/audit_stale_running.py git commit -m "feat(scripts): audit_stale_running for pre-deploy audit (spec §7.3)" ``` --- ### Task C5: Block C 통합 회귀 + 직접 script 처리 (V2 patch P6 + 추가 c) **V2 patch P6 (Codex MINOR #1)**: C5 가 "LLM/image 호출 1회" 검증 의무인데 단순 claim 성공 카운트만 보면 부족 — `_execute()` 자체 호출 횟수도 counter fake 로 검증. **V2 patch 추가 c (Codex IMPORTANT #3 + spec V5 S5)**: `backend/scripts/test_scene_detail.py:122` 가 `original_execute(mode)` 직접 호출 — atomic claim 우회. spec §5.3 의 옵션 A (StepRunner.run 경유 전환) / 옵션 B (명시 quarantine + allowlist) 중 사용자 결정 후 처리. **Files:** - (테스트 추가 없음 — 종합 실행) - Modify or Quarantine: `backend/scripts/test_scene_detail.py` (옵션 A or B, 사용자 결정) **AC:** AC-C1~C12 종합 + AC-C3 - [ ] **Step 1: 직접 script 처리 — 옵션 A or B 결정 (사용자 승인)** | 옵션 | 적용 | 변경 | |---|---|---| | **A. StepRunner.run() 경유 전환** (default 권장) | 일반 운영 path 유지 | `backend/scripts/test_scene_detail.py:122` 의 `result = original_execute(mode)` → `result = runner.run(mode=mode)` 로 교체. fan-out filter 는 `_load_prev_checkpoint` monkeypatch 만으로 충분. `original_*` 변수 제거. | | **B. quarantine** | 개발/디버그 한정 | 파일 상단에 `# QUARANTINED: dev-only — atomic claim 우회. production dispatch 에서 호출 금지` 주석 + docstring. 파일을 `backend/scripts/_quarantined/test_scene_detail.py` 로 이동. AC-C3 grep allowlist 등록. | **default**: 옵션 A. plan 적용 시 사용자 명시 결정 받기. 결정 사항을 본 step 본문에 기록. - [ ] **Step 2: Block C 단위 + 통합 전체** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_try_claim_running.py tests/integration/test_step_runner_run_flow.py tests/integration/test_concurrent_claim.py -v` Expected: 모두 PASS - [ ] **Step 3: 직접 script 우회 검사 (AC-C3, V2 patch P6 grep 재정의)** V2 patch P6: spec §5.3 V5 S5 의 grep 기준 적용 — quarantine allowlist 외 0건 검증. ```bash # 1) 직접 _execute() / original_execute() 호출 검색 (quarantine 제외). # review M2 정정 (Block C closure): rg --glob 은 basename 매칭이라 prefixed # 'backend/scripts/_quarantined/**' 는 cwd 의존. '**/_quarantined/**' 안전. rg "(?:^|[^_])\b(?:original_execute|_execute)\(" backend/scripts/ \ --glob '!**/__pycache__/**' \ --glob '!**/_quarantined/**' ``` Expected: 옵션 A 채택 시 0 lines / 옵션 B 채택 시 allowlist 외 0 lines - [ ] **Step 4: LLM call counter 검증 (V2 patch P6 — Codex MINOR #1)** `tests/integration/test_concurrent_claim.py` 에 추가 — counter fake `_execute` 통합 회귀: ```python def test_concurrent_workers_invoke_execute_only_once(monkeypatch): """V2 patch P6: 두 worker 가 같은 step 잡을 때 _execute() 가 1회만 호출됨. Task C3 의 claim 성공 카운트 검증을 보강 — 실제 _execute() 호출 카운트로 LLM/image 발생 1회만 확인. """ from app.core.step_runner import StepRunner import threading counter = {"calls": 0} counter_lock = threading.Lock() def fake_execute(self, mode="resume"): with counter_lock: counter["calls"] += 1 return {"completed_count": 1, "applicable_count": 1, "failed_count": 0} monkeypatch.setattr(StepRunner, "_execute", fake_execute) # ... two workers race fixture (Task C3 와 동일 setup) ... # 두 worker 중 하나만 _execute 호출 assert counter["calls"] == 1 ``` Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_concurrent_claim.py::test_concurrent_workers_invoke_execute_only_once -v` Expected: PASS - [ ] **Step 5: Block A + B + C 합산 smoke** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/ -v --tb=short 2>&1 | tail -30` Expected: 전체 회귀 0 - [ ] **Step 6: 통계 record** 전체 통과 카운트 capture: ```bash cd backend && PYTHONPATH=. .venv/bin/pytest tests/ -q 2>&1 | tail -5 ``` - [ ] **Step 7: commit (회귀 OK marker)** ```bash # 옵션 A 또는 B 의 script 변경도 함께 stage git add backend/scripts/test_scene_detail.py # 옵션 A 시 # 또는: git add backend/scripts/_quarantined/ # 옵션 B 시 git commit -m "test: Block A+B+C full regression confirmed + script quarantine (24 AC + AC-C3 grep 0)" ``` --- ## Final — 사용자 승인 후 dispatch 재개 ### Task F1: 배포 전 production audit **Files:** (운영 액션 — 코드 변경 없음) - [ ] **Step 1: production DB 에서 audit script 실행** ```bash cd backend && PYTHONPATH=. .venv/bin/python scripts/audit_stale_running.py 2>&1 | tee /tmp/pre_deploy_audit.log ``` - [ ] **Step 2: STALE / PARSE_FAILED / NULL_STARTED row 사용자 결정** audit log 출력 분석. 각 row 별로 사용자가: - "Healthy" → 그대로 - "Stale → 수동 reset" SQL 실행 결정 - "Parse failed / Null started" → manual investigation - [ ] **Step 3: 사용자 명시 OK 받은 후 다음 step 진입** (승인 없이 dispatch 재개 금지) --- ### Task F2: dispatch 재개 **Files:** (운영 액션 — 코드 변경 없음) - [ ] **Step 1: dispatch script 실행** ```bash cd backend && BACKGROUND_RENDER_WORKERS=2 PYTHONPATH=. .venv/bin/python scripts/dispatch_pid_80f62523.py 2>&1 ``` (run_in_background=true) - [ ] **Step 2: 3 monitor 가동** (이전 세션의 monitor 명령 재가동 — `tail -F /tmp/dispatch_80f62523.log` filter, PNG watch, PG advisory watchdog) - [ ] **Step 3: 24시간 모니터링 (spec §7.4)** | 신호 | 임계값 | 조치 | |---|---|---| | `step.resume_blocked` | 시간당 5건+ | timeout 조정 또는 수동 reset 검토 | | `step.verify_crashed` | 시간당 1건+ | 즉시 stack trace 분석 | | `step.already_running` | 시간당 3건+ | claim race 또는 stale steal 미작동 확인 | | running timeout 초과 재발 | 시간당 1건+ | atomic steal SQL 동작 확인 | --- ## Self-Review ### Spec coverage (V2 갱신) - AC-A1~A6 → Block A Tasks A1~A7 ✅ - AC-B1~B9 → Block B Tasks **B0**~B14 ✅ (V2 patch P4 — B0 신규) - AC-C1~C12 → Block C Tasks C1~C5 ✅ - §7.3 audit → Task C4 ✅ - §7.4 monitoring → Task F2 ✅ - §4.7 owner-aware → Task B9 + run() refactor (Task C2) + `_update_step_run_strict` (V2 추가 d) ✅ - §4.8 NOT_APPLICABLE → Task B10 ✅ - §3.5 card hash assert (V5 S1: caller pre-capture 책임) → Task A6 재작성 (V2 patch P2) ✅ - §1.5 step_run schema → Task C1 INSERT 컬럼 명시 ✅ - §3.3 item_id deterministic → Task A2 (build map) + A3 (apply uses item_id) + A5 (schema/prompt, V2 patch P1) ✅ - **V5 S1** caller pre-capture → Task A6 (V2 patch P2) ✅ - **V5 S2** `_execute_and_finalize` 6 단계 보존 → Task B12 + C2 (V2 patch P3) ✅ - **V5 S3** start_step / run_steps_batch 두 path → Task B13 (V2 patch P6) ✅ - **V5 S4** verify_completion fixture → Task B8 재작성 (V2 patch P5) ✅ - **V5 S5** script quarantine → Task C5 신규 step (V2 추가 c) ✅ - **V5 S6** AppError extra 금지 → Task C2 본문 정정 (V2 추가 a) + grep 검증 ✅ - **V5 S7** `_get_step_run` dict shape → Task **B0** 신규 (V2 patch P4) ✅ ### Placeholder scan - TODO/TBD 0건 ✅ - 모든 step 에 실제 code/command 포함 ✅ - 파일 경로 절대 line number 명시 (예: `backend/app/core/step_runner.py:546`) ✅ ### Type consistency - `ResumeAction` enum value 6개 통일 (skip/rerun_self/force_explicit/stale_running_recovery/block/not_applicable) ✅ - `ResumeDecision(action, reason, origin, expected_started_at, expected_run_id)` 모든 호출자에서 일관 ✅ - `_try_claim_running(allow_stale_steal, expected_started_at, expected_run_id)` 시그니처 spec/plan 일치 ✅ - `_update_step_run(require_owner: bool = False) -> bool` + `_update_step_run_strict(...)` (V2 추가 d) 일관 ✅ - `_get_step_run() -> Optional[Dict[str, Any]]` (V2 patch P4 / V5 S7 — dict shape) ✅ - `_evaluate_*` 메서드 모두 ResumeDecision 반환 ✅ - `_execute_rerun_self` / `_execute_force` / `_execute_and_finalize` 시그니처 (V2 patch P3 / V5 S2) ✅ ### V1 self-review 의 false positive 정정 (Codex NEEDS_REVISION 반영) - ~~AC-A3 → Task A6 (`_assert_card_hash_unchanged`)~~ — v1 의 매핑은 caller pre-capture 빠뜨려 미충족. **v2 정정**: V5 S1 + V2 P2 로 caller 의 `_capture_card_hash_state` 책임 명시. - ~~AC-B7 → run-all 동일성 회귀~~ — v1 의 매핑은 같은 runner 두 번 호출이라 두 path 미검증. **v2 정정**: V5 S3 + V2 P6 로 start_step / run_steps_batch 양쪽 stub. - ~~AC-C3 → 우회 0~~ — v1 의 grep 은 backend/scripts/test_scene_detail.py 잡힘. **v2 정정**: V5 S5 + V2 추가 c 로 옵션 A/B + allowlist grep 재정의. ### V2 의 후행 결함 정정 (V2.1 patch — Codex v2 NEEDS_REVISION 반영) - **BLOCKING #1**: V2 의 B0 가 dict shape 만 허용했는데 B11 에서 `isinstance(existing, tuple)` fallback 잔존 + fixture 가 tuple 사용. **v2.1 정정**: B11 의 fallback 코드 제거 + fixture 모두 dict 변환. 검증 grep `existing\[\d\]` 0 lines. - **BLOCKING #2**: V2 의 `_execute_and_finalize` 가 exit verify AppError / `_execute()` exception 시 status='failed' 기록 누락 → 'running' status leak. **v2.1 정정**: exit verify 주변 try/except + outer try/except (AppError + Exception 양쪽) 추가. 모든 path 에서 `_update_step_run_strict("failed")` 후 re-raise (owner_lost silent absorb). - **BLOCKING #3**: V2 의 B8 fixture 가 `step._load_chain_bg_owned_by_shot` / `_g41_render_prompt_card` patch — 실제 코드는 `SceneContextLoader(self)._load_chain_bg_owned_by_shot()` 객체 메서드 + `from app.core.steps.render_prompt_card import build_render_prompt_card as _g41_build_card` 함수 내부 import. **v2.1 정정**: target 모두 source module 의 정확 위치로 재설정. - **IMPORTANT #1**: V2 의 B8 step 3 가 early-return 패턴으로 분류 — 현 코드의 aggregation (`missing_msgs.extend` + severity 계산) 구조 손상 위험. **v2.1 정정**: aggregation 보존 + 단일 return 자리에 origin priority mirror 만 추가 명시. - **IMPORTANT #2**: V2 의 A6 expected 가 "3 passed" 인데 V2 P2 가 1 test 추가. **v2.1 정정**: 4 passed 로 정합. ### V2.1 검증 grep 의무 (commit 전) V2.1.1 patch (Codex NEEDS_SMALL_PATCH IMPORTANT): broad grep (`step\._load_chain_bg`, `_g41_render_prompt_card`) 은 self-review 의 BLOCKING #3 설명 본문 + 과거 G4.x plan 들이 매치 → 0 lines 불가능. **잘못된 patch 패턴만** 정밀 grep 으로: ```bash # B11 dict-only contract — code 에서 검증 rg "existing\[\d\]" backend/app/core/step_runner.py # 0 lines (B0 grep) rg "isinstance\(existing, tuple\)" backend/ # 0 lines # B8 잘못된 patch target — fixture 의 monkeypatch / patch 호출에서만 검증 (V2.1.1 정밀화) rg 'monkeypatch\.setattr\([^\n]*step\._load_chain_bg' backend/tests/ # 0 lines (instance method 가 아닌 SceneContextLoader 메서드만 patch) rg 'patch\("app\.core\.steps\.detail_steps\._g41_render_prompt_card"' backend/tests/ # 0 lines (옛 alias 잔존 X) rg 'step\._g41_render_prompt_card' backend/tests/ # 0 lines (instance method 도 아님) # AppError extra (V5 S6 mirror) rg "AppError\(.*extra=" backend/ # 0 lines ``` ### Test 회귀 baseline 명시 - 각 task 끝에 `pytest -k ` 회귀 검증 - Block A/B/C 끝에 광범위 smoke (`pytest tests/ -v`) - V2 patch 추가 검증 — Block A B0 신규 (`test_get_step_run_shape.py`) + B8 fixture 기반 (`test_scene_detail_verify_origin.py`) + B12 finalize 6 단계 (`test_resume_decision.py`) + B13 두 path (`test_resume_single_vs_run_all_consistency.py`) + C5 LLM counter (`test_concurrent_claim.py`) --- ## Patch History | Date | Version | 작성자 | 비고 | |---|---|---|---| | 2026-05-07 | v1 | Claude (Opus 4.7) | spec v4 (DRAFT_REVIEWED_PATCHED_V4) 기반 — Block A/B/C 26 task 분해 | | 2026-05-08 | v2 | Claude (Opus 4.7) | Codex NEEDS_REVISION (BLOCKING 6 + IMPORTANT 4 + MINOR 2) 반영 — 11 patch (P1 prompt path / P2 pre-capture / P3 finalize 보존 + AppError extra / P4 신규 B0 dict shape / P5 verify_completion fixture / P6 두 path + LLM counter / 추가 a-e). spec v5 (DRAFT_REVIEWED_PATCHED_V5) 기반. False positive 매핑 3건 (AC-A3/B7/C3) 정정. | | 2026-05-08 | v2.1 | Claude (Opus 4.7) | Codex v2 NEEDS_REVISION (BLOCKING 3 + IMPORTANT 2) 반영 — (1) B11 의 tuple fallback 제거 + fixture dict 통일 (BLOCKING #1) / (2) `_execute_and_finalize` exit verify crash + execute exception catch → status='failed' 기록 (BLOCKING #2) / (3) B8 patch target 정정 — `SceneContextLoader._load_chain_bg_owned_by_shot` + `app.core.steps.render_prompt_card.build_render_prompt_card` (BLOCKING #3) / (4) verify_completion aggregation 보존 + origin priority mirror 명시 (IMPORTANT #1) / (5) A6 expected 3 → 4 passed 정합 (IMPORTANT #2). Spec 추가 변경 없음. | | 2026-05-08 | v2.1.1 | Claude (Opus 4.7) | Codex v2.1 NEEDS_SMALL_PATCH 반영 — (1) `_execute_and_finalize` 의 `failed_recorded` local flag 추가: inner exit verify catch 가 specific `error_message` 로 기록한 status='failed' 를 outer except 의 generic message ("AppError in _execute_and_finalize") 가 같은 owner 로 덮는 race 차단 (BLOCKING) / (2) V2.1 검증 grep 정밀화 — broad `step\._load_chain_bg` / `_g41_render_prompt_card` 는 self-review 본문 + 과거 G4.x plan 매치라 0 lines 불가 → fixture 의 잘못된 patch call 만 잡는 정밀 grep 으로 변경 (IMPORTANT). Spec 추가 변경 없음. | | 2026-05-08 | v2.1.2 | Claude (Opus 4.7) | Block A T3/T5 실행 중 발견된 plan-level lessons — (a) T3 (`_apply_scene_fixes` caller 전환) 와 T5 (schema/prompt/producer item_id propagation) 가 **atomic** 이어야 함 (T3 만 commit 시 production 에서 `t2i_review.missing_item_id` raise 폭주, Codex T3 review 에서 발견). Block A intro 에 atomic dependency + immediate sequence 가드 추가. (b) T5 schema 형식 deviation 명시 — plan A5 의 `fixes.items` (issue-level) 가 producer 의 `results[].issues[]` two-tier 와 mismatch → **result-level `item_id` required** 채택 (commit 602e44b). 향후 implementer 가 plan 본문 그대로 적용 X. (c) T5 v2 stabilization (commit 764d501) — `result.item_id` 누락 / `SchemaValidationError` / `failed_batches > 0` 모두 fail-fast (AppError raise). `logger.warning + continue` 패턴은 `feedback_no_silent_fallback.md` 정책 위반. integration test 3건 raise 가드. Spec 추가 변경 없음. | | 2026-05-08 | v2.1.3 | Claude (Opus 4.7) | Codex Block A 누적 review NEEDS_REVISION (BLOCKING 2 + IMPORTANT 2) 반영 — (B1) `_save_checkpoint_data` 의 manifest missing/unreadable 시 logger.warning+return → AppError(t2i_review.checkpoint_save_failed) raise (silent fallback 차단 — mutator 성공 오판 방지). (B2) `_review_entity_t2i` 의 batch except — T5 v2 와 동일 fail-fast 정책 (SchemaValidationError/AppError re-raise + transient 도 AppError(t2i_review.entity_batch_failed) raise). 옛 `return []` 은 검수 실패를 "문제 없음" 으로 변환. (I1) T7 full-cycle test 에 실제 SceneDetailStep.verify_completion() 호출 fixture 추가 — hash equality only test 는 helper unit 으로 보강 유지. evidence gap 차단. (I2) Task A5 본문 schema 예시 + 검증 script 를 result-level item_id 로 정정 (v2.1.1 의 `fixes.items` 잔재 폐기). intro 가드와 본문 일치 — future implementer 함정 차단. Spec 추가 변경 없음. |