"""#77-A — config drift 명시 승인(step_config_drift_ack) 갈래. config_hash mismatch 는 기본 BLOCK(사용자 의도 확인) 유지. 사용자가 env 로 그 스텝을 명시 승인한 경우에만 비파괴 resume 재실행(RERUN_SELF) — origin 은 runner 의 어느 force 격상 갈래에도 안 걸리는 "config_drift_ack" 라 mode=resume 그대로 떨어진다(step_runner :1160 fall-through). """ from unittest.mock import patch def _make_step(step_id="scene_image_pipeline", cp=None): from app.core.steps.image_steps import SceneImagePipelineStep step = SceneImagePipelineStep.__new__(SceneImagePipelineStep) step.step_id = step_id step.load_checkpoint = lambda: cp return step def test_no_ack_blocks_config_drift(): from app.core.step_runner import ResumeAction step = _make_step(cp={}) with patch("app.core.config.settings.step_config_drift_ack", ""): decision = step._evaluate_contract_drift( "config_hash mismatch: 체크포인트=aaaa, 현재=bbbb") assert decision.action == ResumeAction.BLOCK def test_ack_allows_nondestructive_rerun(): from app.core.step_runner import ResumeAction step = _make_step(cp={}) with patch("app.core.config.settings.step_config_drift_ack", "scene_image_pipeline"): decision = step._evaluate_contract_drift( "config_hash mismatch: 체크포인트=aaaa, 현재=bbbb") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "config_drift_ack" def test_ack_for_other_step_still_blocks(): from app.core.step_runner import ResumeAction step = _make_step(cp={}) with patch("app.core.config.settings.step_config_drift_ack", "entity_t2i, scene_detail"): decision = step._evaluate_contract_drift( "config_hash mismatch: 체크포인트=aaaa, 현재=bbbb") assert decision.action == ResumeAction.BLOCK def test_ack_does_not_cover_schema_version_mismatch(): # 승인은 config_hash 전용 — schema mismatch 는 기존 allowlist 정책 그대로. from app.core.step_runner import ResumeAction step = _make_step(cp={}) with patch("app.core.config.settings.step_config_drift_ack", "scene_image_pipeline"): decision = step._evaluate_contract_drift( "schema_version mismatch: 체크포인트=1, 현재=2") assert decision.action == ResumeAction.BLOCK def test_public_run_force_latch_blocks_before_any_mutation(tmp_path): # Codex 3차 리뷰 BLOCK: _execute 첫 문장 검사만으로는 public # run(force) 이 그 전에 cleanup→invalidate_downstream→ # clear_checkpoint 를 밟는다(step_runner._execute_force). 래치는 # validate_mode(run :999 — 모든 변형·claim 전)가 막고, 스파이 0회로 # "무접촉"을 잠근다. from unittest.mock import MagicMock from app.services.still_recipe_service import ( JIT_LATCH_FILENAME, StillJitRegenLimitExceeded, ) import pytest pid, eid = "SAMPLE_P", "SAMPLE_E" recipe = tmp_path / pid / "images" / eid / "scene" / "recipe" recipe.mkdir(parents=True) (recipe / JIT_LATCH_FILENAME).write_text("{}", encoding="utf-8") step = _make_step() step.project_id, step.episode_id = pid, eid step.check_gate = MagicMock() step.check_applicability = MagicMock(return_value=True) step.cleanup_artifacts = MagicMock() step.invalidate_downstream = MagicMock() step.clear_checkpoint = MagicMock() step._execute = MagicMock() with patch("app.core.config.settings.projects_dir", str(tmp_path)): for mode in ("force", "resume"): with pytest.raises(StillJitRegenLimitExceeded): step.run(mode=mode) step.cleanup_artifacts.assert_not_called() step.invalidate_downstream.assert_not_called() step.clear_checkpoint.assert_not_called() step._execute.assert_not_called() def test_step_entry_latch_blocks_all_modes(tmp_path): # Codex 재재리뷰 BLOCK 2: 래치는 _execute 최앞단 — 선행 유료 구간 # (world guide·T2I fallback)과 force 의 recipe 아카이브(래치까지 # 치우는 우회)보다 먼저 막는다. __new__ 스텁(tracer 준비 없음)에서도 # 래치 예외가 나는 것 자체가 "무엇보다 먼저"의 증명이다. from app.services.still_recipe_service import ( JIT_LATCH_FILENAME, StillJitRegenLimitExceeded, ) import pytest pid, eid = "SAMPLE_P", "SAMPLE_E" recipe = tmp_path / pid / "images" / eid / "scene" / "recipe" recipe.mkdir(parents=True) (recipe / JIT_LATCH_FILENAME).write_text("{}", encoding="utf-8") step = _make_step() step.project_id, step.episode_id = pid, eid with patch("app.core.config.settings.projects_dir", str(tmp_path)): for mode in ("resume", "force"): with pytest.raises(StillJitRegenLimitExceeded): step._execute(mode=mode) def test_step_entry_latch_respects_switch_off(tmp_path): # 되돌림 레버 off = 래치 검사도 예전처럼 없음 — 래치 예외가 아닌 # 다른 지점(스텁이라 tracer 준비 없음)까지 진행된다. from app.services.still_recipe_service import ( JIT_LATCH_FILENAME, StillJitRegenLimitExceeded, ) import pytest pid, eid = "SAMPLE_P", "SAMPLE_E" recipe = tmp_path / pid / "images" / eid / "scene" / "recipe" recipe.mkdir(parents=True) (recipe / JIT_LATCH_FILENAME).write_text("{}", encoding="utf-8") step = _make_step() step.project_id, step.episode_id = pid, eid with patch("app.core.config.settings.projects_dir", str(tmp_path)), \ patch("app.core.config.settings.still_jit_verify_enabled", False, create=True): with pytest.raises(Exception) as ei: step._execute(mode="resume") assert not isinstance(ei.value, StillJitRegenLimitExceeded) def test_target_scope_drift_still_wins_over_ack(): # base 동일(표적 목록만 변경)은 승인과 무관하게 기존 표적 drift 경로. from app.core.step_runner import ResumeAction step = _make_step(cp={"target_scope_base_hash": "samebase"}) step._config_hash_base = lambda: "samebase" with patch("app.core.config.settings.step_config_drift_ack", ""): decision = step._evaluate_contract_drift( "config_hash mismatch: 체크포인트=aaaa, 현재=bbbb") assert decision.action == ResumeAction.RERUN_SELF assert decision.origin == "contract_drift" assert "target-scope" in decision.reason def test_the_base_runner_does_not_open_on_ack_for_a_non_image_step(): """★2026-09-03 Codex BLOCK: 범용 ack 로 RERUN_SELF 를 열면 실제 LLM 을 다시 부르는 스텝(entity_detail :1240)의 산출이 달라져도 하류를 보존한다 — 계보가 끊긴다. 승인 갈래는 샷별 JIT 지문 검증(#77-B)이 있는 image step override 에만 있다. hash 조리법만 바뀐 경우는 재실행이 아니라 **정확한 hash adoption**(tools/grounding_audit/canary_hash_adoption)이다.""" from app.core.step_runner import ResumeAction from app.core.steps.entity_steps import EntityDetailStep step = EntityDetailStep.__new__(EntityDetailStep) step.step_id = "entity_detail" step.load_checkpoint = lambda: {} with patch("app.core.config.settings.step_config_drift_ack", "entity_detail"): d = step._evaluate_contract_drift("config_hash mismatch: 체크포인트=aaaa, 현재=bbbb") assert d.action == ResumeAction.BLOCK and d.origin == "contract_drift"