"""Block B T1~T3 — `ResumeAction` enum + `ResumeDecision` dataclass + `StepRunner._evaluate_resume_decision()` decision helper. Plan v2.1.3 / spec V5 §2.3: - 6 ResumeAction 분기 (skip / rerun_self / force_explicit / stale_running_recovery / block / not_applicable). - `ResumeDecision` 5 필드 (action / reason / origin / expected_started_at / expected_run_id) — STALE_RUNNING_RECOVERY claim atomic steal 의존. - 본 commit (T1~T3) scope: type 도입 + helper structural skeleton 만. 실제 정책 변경 (running 자동 force 금지 / verify_crashed origin 분리 / rerun_self vs force 분리) 은 후속 task (B4 / B7 / B12). """ from __future__ import annotations from typing import Optional from unittest.mock import MagicMock import pytest # --------------------------------------------------------------------------- # Type tests (plan B3 AC-B1) # --------------------------------------------------------------------------- class TestResumeActionEnum: def test_action_enum_has_six_values(self): from app.core.step_runner import ResumeAction assert {a.value for a in ResumeAction} == { "skip", "rerun_self", "force_explicit", "stale_running_recovery", "block", "not_applicable", } def test_action_enum_members_uppercase(self): from app.core.step_runner import ResumeAction assert ResumeAction.SKIP.value == "skip" assert ResumeAction.RERUN_SELF.value == "rerun_self" assert ResumeAction.FORCE_EXPLICIT.value == "force_explicit" assert ResumeAction.STALE_RUNNING_RECOVERY.value == "stale_running_recovery" assert ResumeAction.BLOCK.value == "block" assert ResumeAction.NOT_APPLICABLE.value == "not_applicable" class TestResumeDecisionDataclass: def test_decision_default_optional_fields(self): from app.core.step_runner import ResumeAction, ResumeDecision 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(self): from app.core.step_runner import ResumeAction, ResumeDecision 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" def test_decision_is_frozen(self): from app.core.step_runner import ResumeAction, ResumeDecision from dataclasses import FrozenInstanceError d = ResumeDecision(action=ResumeAction.SKIP, reason="x") with pytest.raises(FrozenInstanceError): d.action = ResumeAction.BLOCK # --------------------------------------------------------------------------- # `_evaluate_resume_decision()` helper tests # # helper 의 직접 부수효과는 0 — `_record_recovery` / `_update_step_run` / # `logger.warning` / `raise` 호출 없음. recovery / log / AppError 는 caller # (run()) 책임. # # 단, `load_checkpoint` (archive fallback rewrite) 와 `_safe_verify_completion` # (verify crash logger.error 흡수) 의 기존 side effect 는 보존 — strict-pure 가 # 아님. Block C (claim 전 non-mutating decision) 전제 와의 잠재 충돌은 후속 # task (B7/B8 verify_crashed origin) 에서 정리. # --------------------------------------------------------------------------- def _make_runner_stub( *, step_id: str = "text_cleanup", existing: Optional[dict] = None, cp: Optional[dict] = None, cp_mismatch: Optional[str] = None, verify_complete: bool = True, verify_missing: Optional[list] = None, verify_origin: Optional[str] = None, verify_raises: Optional[Exception] = None, project_config: Optional[dict] = None, ): """StepRunner instance + 모든 helper input 을 monkeypatch. Block B B11 (plan v2.1.3): verify_origin / verify_raises 추가 — helper 가 verify report.origin 을 디스패치하는지 검증. """ from app.core.integrity_report import CompletionReport from app.core.step_runner import StepRunner runner = StepRunner.__new__(StepRunner) runner.step_id = step_id runner.project_config = project_config or {} runner._get_step_run = lambda sid: existing runner.load_checkpoint = lambda: cp runner._check_cp_mismatch = lambda c: cp_mismatch def _verify(): if verify_raises is not None: raise verify_raises kwargs = { "is_complete": verify_complete, "missing": verify_missing or [], "severity": "clean" if verify_complete else "missing", "metadata": {}, } # B1 origin 필드 — caller 가 명시한 경우에만 set (default 'artifact_missing') if verify_origin is not None: kwargs["origin"] = verify_origin return CompletionReport(**kwargs) runner._safe_verify_completion = _verify return runner class TestDecisionHelperForceMode: def test_mode_force_returns_force_explicit(self): from app.core.step_runner import ResumeAction runner = _make_runner_stub() decision = runner._evaluate_resume_decision(mode="force") assert decision.action == ResumeAction.FORCE_EXPLICIT assert "force" in decision.reason.lower() class TestDecisionHelperFirstRun: def test_first_run_no_step_run_row_returns_rerun_self(self): """existing == None → RERUN_SELF (origin=None — first-run 마커).""" from app.core.step_runner import ResumeAction runner = _make_runner_stub(existing=None) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin is None # first-run marker (no cleanup, no recovery side-effect) class TestDecisionHelperCompletedClean: def test_completed_with_clean_cp_and_verify_passes_returns_skip(self): from app.core.step_runner import ResumeAction existing = { "status": "completed", "run_id": "r1", "started_at": "2026-05-08T00:00:00", "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": "2026-05-08T00:01:00", } cp = {"schema_version": 1, "config_hash": "h1", "data": {}} runner = _make_runner_stub( existing=existing, cp=cp, cp_mismatch=None, verify_complete=True, ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.SKIP class TestDecisionHelperCompletedMismatch: def test_completed_with_cp_none_returns_rerun_self_artifact_missing(self): """status=completed + cp=None → RERUN_SELF (artifact_missing). D1 fix (PID 0bb48ebf 사고 패턴) — silent skip 차단. helper 는 origin ='artifact_missing' 으로 표시 — caller (run()) 가 record_recovery 호출. """ from app.core.step_runner import ResumeAction existing = {"status": "completed", "run_id": "r1", "started_at": None, "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": None} runner = _make_runner_stub(existing=existing, cp=None) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "artifact_missing" assert "checkpoint" in decision.reason.lower() or "cp" in decision.reason.lower() def test_completed_with_cp_mismatch_returns_block_after_b5(self): """B5 (plan v2.1.3 §4.5): schema/config_hash mismatch → BLOCK + origin= contract_drift. 이전 RERUN_SELF + artifact_missing 정책은 reverse. entity_t2i allowlist 케이스는 test_legacy_schema_bump_allowlist.py 참조. """ from app.core.step_runner import ResumeAction existing = {"status": "completed", "run_id": "r1", "started_at": None, "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": None} cp = {"schema_version": 1, "config_hash": "old", "data": {}} runner = _make_runner_stub( existing=existing, cp=cp, cp_mismatch="config_hash mismatch: project_config 변경 감지", ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.BLOCK assert decision.origin == "contract_drift" assert "config_hash" in decision.reason def test_completed_with_verify_failure_returns_rerun_self(self): from app.core.step_runner import ResumeAction existing = {"status": "completed", "run_id": "r1", "started_at": None, "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": None} cp = {"schema_version": 1, "config_hash": "h1", "data": {}} runner = _make_runner_stub( existing=existing, cp=cp, cp_mismatch=None, verify_complete=False, verify_missing=["sentinel_drift"], ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "artifact_missing" assert "verify" in decision.reason.lower() class TestDecisionHelperStrictResume: def test_completed_mismatch_with_strict_resume_returns_block(self): """strict_resume=True + cp not None + mismatch → BLOCK (legacy 보존). legacy strict 동작 — caller 가 AppError(step.resume_invalid) raise. """ from app.core.step_runner import ResumeAction existing = {"status": "completed", "run_id": "r1", "started_at": None, "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": None} cp = {"schema_version": 1, "config_hash": "old", "data": {}} runner = _make_runner_stub( existing=existing, cp=cp, cp_mismatch="config_hash mismatch", project_config={"strict_resume": True}, ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.BLOCK assert decision.origin == "strict_resume" # caller dispatch marker assert "config_hash" in decision.reason # raw mismatch carried for AppError message def test_strict_resume_with_cp_none_falls_back_to_rerun_self(self): """strict_resume + cp=None → BLOCK 미적용 (legacy: cp 없으면 자동 force). 현재 run() line 593-604 동작 보존: `strict_resume 은 cp 있을 때만 의미`. """ from app.core.step_runner import ResumeAction existing = {"status": "completed", "run_id": "r1", "started_at": None, "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": None} runner = _make_runner_stub( existing=existing, cp=None, project_config={"strict_resume": True}, ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "artifact_missing" class TestDecisionHelperPriorState: """status in {failed / partial / stale / pending} — RERUN_SELF + origin= 'prior_state' (force-like recovery 자연 진행). NOTE: 'running' 은 Block B B4 에서 분리 — `_evaluate_running_state` 가 timeout 기반으로 BLOCK 또는 STALE_RUNNING_RECOVERY 결정. test 는 test_running_state_evaluation.py 참조. """ @pytest.mark.parametrize("status", [ "failed", "partial", "stale", "pending", ]) def test_non_completed_status_returns_rerun_self_prior_state(self, status): from app.core.step_runner import ResumeAction existing = {"status": status, "run_id": "r1", "started_at": "2026-05-08", "completed_count": 0, "applicable_count": 5, "recovery_count": 0, "updated_at": "2026-05-08"} runner = _make_runner_stub(existing=existing) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "prior_state" assert status in decision.reason class TestDecisionHelperUnknownStatus: def test_unknown_status_returns_rerun_self_first_run_marker(self): """현재 run() 은 unknown status 시 fall through → execute (cleanup 없음). helper 는 동일 동작 보존을 위해 origin=None 으로 표시 — caller 가 first-run 처럼 cleanup 없이 execute. 후속 task (B11) 에서 BLOCK 정책 도입. """ from app.core.step_runner import ResumeAction existing = {"status": "weird_status", "run_id": "r1", "started_at": None, "completed_count": 0, "applicable_count": 0, "recovery_count": 0, "updated_at": None} runner = _make_runner_stub(existing=existing) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin is None # first-run-like — no cleanup, no recovery # --------------------------------------------------------------------------- # Side-effect-free invariant — observation only # --------------------------------------------------------------------------- class TestDecisionHelperPurity: """direct 부수효과 가드 — `_record_recovery` / `_update_step_run` / raise 0. NOTE: load_checkpoint / _safe_verify_completion 의 기존 side effect 는 별도 — Block C (claim 전 non-mutating) 전제는 B7/B8 에서 정리. """ def test_helper_does_not_call_record_recovery(self): """helper 는 _record_recovery / _check_recovery_exhausted 호출 X. recovery / update / raise 디스패치는 caller (run()) 책임. recovery counting 누락 시 silent skip 사고 재발 — 가드 의무. """ existing = {"status": "completed", "run_id": "r1", "started_at": None, "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": None} runner = _make_runner_stub(existing=existing, cp=None) # mock 으로 호출 추적 runner._record_recovery = MagicMock() runner._check_recovery_exhausted = MagicMock() runner._evaluate_resume_decision(mode="resume") assert not runner._record_recovery.called assert not runner._check_recovery_exhausted.called def test_helper_does_not_raise_app_error_for_strict_resume(self): """strict_resume + mismatch 시 BLOCK decision 반환 — raise X. caller (run()) 가 BLOCK 디스패치 시 AppError raise 하면서 message 형식 보존. """ from app.core.step_runner import ResumeAction existing = {"status": "completed", "run_id": "r1", "started_at": None, "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": None} cp = {"schema_version": 1, "config_hash": "old", "data": {}} runner = _make_runner_stub( existing=existing, cp=cp, cp_mismatch="config_hash mismatch", project_config={"strict_resume": True}, ) # 예외 없이 decision 반환 decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.BLOCK # --------------------------------------------------------------------------- # Block B B11 — verify report.origin 정책 분기 # # B7 / B8 가 origin 분류를 도입한 후, helper 가 report.origin 을 소비해 # contract_drift / invariant_drift → BLOCK, artifact_missing 등 → RERUN_SELF # 디스패치. cp_mismatch 경로 (schema/config_hash) 는 별도 (B5 follow-up) — 본 # commit scope 밖. # --------------------------------------------------------------------------- def _completed_existing() -> dict: return { "status": "completed", "run_id": "r1", "started_at": "2026-05-08T00:00:00", "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": "2026-05-08T00:01:00", } class TestDecisionHelperVerifyOriginContractDrift: def test_completed_verify_contract_drift_returns_block(self): """V8 verify 가 contract_drift origin 으로 is_complete=False 반환 → helper 가 BLOCK 디스패치 (auto force-like recovery 차단).""" from app.core.step_runner import ResumeAction cp = {"schema_version": 1, "config_hash": "h1", "data": {}} runner = _make_runner_stub( existing=_completed_existing(), cp=cp, cp_mismatch=None, verify_complete=False, verify_origin="contract_drift", verify_missing=["contract: loader violation"], ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.BLOCK assert decision.origin == "contract_drift" # caller 가 AppError message 포맷 시 사용 — report.missing 또는 contract 신호 assert "contract" in decision.reason.lower() or "loader" in decision.reason.lower() class TestDecisionHelperVerifyOriginInvariantDrift: def test_completed_verify_invariant_drift_returns_block(self): """V8 verify 가 invariant_drift origin 으로 is_complete=False 반환 → helper 가 BLOCK 디스패치 (1차 정책 — D4 까지 임시). sentinel/card hash drift 같이 mutator 또는 수동 편집 가능성 — 자동 recovery 시 사고 위험. 사용자 명시 force 요구. """ from app.core.step_runner import ResumeAction cp = {"schema_version": 1, "config_hash": "h1", "data": {}} runner = _make_runner_stub( existing=_completed_existing(), cp=cp, cp_mismatch=None, verify_complete=False, verify_origin="invariant_drift", verify_missing=["3 owned_validation sentinel_drifted: ..."], ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.BLOCK assert decision.origin == "invariant_drift" assert "drift" in decision.reason.lower() or "sentinel" in decision.reason.lower() class TestDecisionHelperVerifyOriginArtifactMissing: def test_completed_verify_artifact_missing_returns_rerun_self(self): """V8 verify 가 artifact_missing origin (default) 반환 → helper 가 기존 RERUN_SELF + recovery 경로 유지 (DB row / PNG / cp 부재 = 재실행 안전).""" from app.core.step_runner import ResumeAction cp = {"schema_version": 1, "config_hash": "h1", "data": {}} runner = _make_runner_stub( existing=_completed_existing(), cp=cp, cp_mismatch=None, verify_complete=False, verify_origin="artifact_missing", verify_missing=["scene_image PNG missing for shot 1"], ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "artifact_missing" def test_completed_verify_default_origin_returns_rerun_self(self): """origin 미지정 (B1 default 'artifact_missing') → RERUN_SELF (backward-compat).""" from app.core.step_runner import ResumeAction cp = {"schema_version": 1, "config_hash": "h1", "data": {}} runner = _make_runner_stub( existing=_completed_existing(), cp=cp, cp_mismatch=None, verify_complete=False, verify_origin=None, # default → artifact_missing verify_missing=["something missing"], ) decision = runner._evaluate_resume_decision(mode="resume") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "artifact_missing" class TestDecisionHelperVerifyCrashedPropagation: def test_verify_crashed_app_error_propagates_through_helper(self): """B7 의 _safe_verify_completion 이 step.verify_crashed AppError raise → helper 가 catch 안 함. caller (run() / step_execution_service) 가 fail-fast. helper 의 직접 부수효과는 0 이지만 verify path 의 raise 는 보존 (Block C non-mutating 전제 조정은 후속 — 본 commit 은 propagation 만 검증). """ from app.core.errors import AppError from app.core.step_runner import StepRunner cp = {"schema_version": 1, "config_hash": "h1", "data": {}} runner = _make_runner_stub( existing=_completed_existing(), cp=cp, cp_mismatch=None, verify_raises=AppError( code="step.verify_crashed", message="card recompute crashed: KeyError: ...", ), ) with pytest.raises(AppError) as exc_info: runner._evaluate_resume_decision(mode="resume") assert exc_info.value.code == "step.verify_crashed" assert "card recompute" in exc_info.value.message class TestDecisionExemptTagsOneShotContract: """Codex 재리뷰 HIGH (2026-08-12) — completed 경계를 decision 층에서 고정. 운영자 무콘티 예외 태그는 one-shot override 라 config hash 에 접히지 않는다(tests/core/test_shot_conti_light_lane.py:: test_exempt_tags_stay_out_of_config_hash 가 hash 불변을 잠근다). hash 불변 → cp_mismatch 없음 → 이 시험이 잠그는 completed+clean+ verify pass 경로에서 태그를 어떻게 바꿔 두든 SKIP. 두 시험의 연역 결합으로 "완료 후 태그 add/remove 는 조용히 무시되며, 반영은 명시 force/invalidate 뿐"이라는 운영 계약이 완결된다. """ def _skip_runner(self): existing = { "status": "completed", "run_id": "r1", "started_at": "2026-05-08T00:00:00", "completed_count": 1, "applicable_count": 1, "recovery_count": 0, "updated_at": "2026-05-08T00:01:00", } cp = {"schema_version": 1, "config_hash": "h1", "data": {}} return _make_runner_stub( existing=existing, cp=cp, cp_mismatch=None, verify_complete=True, ) def test_completed_skip_regardless_of_exempt_tag_state(self): from unittest.mock import patch from app.core.step_runner import ResumeAction for tags in ("", "S75sh7", "S75sh7,S3sh1"): with patch( "app.core.config.settings.bgfirst_no_conti_exempt_tags", tags, create=True, ): decision = self._skip_runner()._evaluate_resume_decision( mode="resume") assert decision.action == ResumeAction.SKIP, ( f"태그={tags!r} 에서도 completed 는 SKIP 이어야 한다" ) class TestTheDecisionCarriesThePreClaimRow: """★claim 뒤 running 이 아니라 claim 전 상태가 판정에 실린다 (Codex 2026-09-03 07:35 · 실측 f7cc45c576c0).""" def test_prior_fields_come_from_the_row_read_before_claim(self): from app.core.step_runner import ResumeAction existing = {"status": "partial", "run_id": "r-old", "updated_at": "2026-09-02T22:26:31", "started_at": None, "completed_count": 8, "applicable_count": 8, "recovery_count": 0} runner = _make_runner_stub(existing=existing, cp={"config_hash": "h"}, verify_complete=True) d = runner._evaluate_resume_decision(mode="resume") assert d.action == ResumeAction.RERUN_SELF and d.origin == "prior_state" assert d.prior_status == "partial" and d.prior_run_id == "r-old" and d.prior_updated_at == "2026-09-02T22:26:31" def test_no_row_means_no_prior(self): runner = _make_runner_stub(existing=None) d = runner._evaluate_resume_decision(mode="resume") assert d.prior_status is None and d.prior_run_id is None