"""canary 가 배경 모드를 **손으로 적지 않고** process 의 실제 설정에서 읽는가. ★실측 (2026-09-02): `fixture_config` 가 `background_mode: True` 를 손으로 적어 두었다. 그러면 process 가 `BACKGROUND_MODE=off` 로 떠도 계획표는 배경 사슬을 세고, 켜져 있으면 이미지 문(`floor_plan_render` = gpt-image-2)이 계획에 없는 채로 나간다. 이제 ①술어에 묻고 ②시나리오의 넷째 축으로 적고 ③선언과 실제가 다르면 서고 ④같은 run 에서 설정을 뒤집을 때 끝난 CP 가 그 설정을 `config_hash` 에 접으면 선다 (Codex 조건 2026-09-02 · 내가 코드로 확인). """ from __future__ import annotations import json import pytest from app.core.config import settings from tools.grounding_audit import canary_run as cr @pytest.fixture def bg(monkeypatch): def _set(value: str): monkeypatch.setattr(settings, "background_mode", value) return _set class TestTheDeclarationAsksThePredicate: def test_off_means_false(self, bg): bg("off") assert cr.fixture_config("v2_chunk")["background_mode"] is False assert cr.background_actual() == "off" def test_on_means_true(self, bg): bg("on") assert cr.fixture_config("v2_chunk")["background_mode"] is True assert cr.background_actual() == "on" def test_the_legacy_alias_counts_as_on(self, bg): bg("floor_plan_anchored") assert cr.background_actual() == "on" class TestTheScenarioCarriesTheRealSetting: def test_it_records_the_actual_value(self, bg): bg("off") sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") assert sc["background"] == "off" assert "background" in cr._SCENARIO_AXES def test_a_declaration_that_matches_passes(self, bg): bg("off") sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide", background="off") assert sc["background"] == "off" def test_a_declaration_that_lies_is_refused(self, bg): """★양성 대조 — `--background off` 라 해 놓고 process 가 on 이면 선다.""" bg("on") with pytest.raises(cr.ScopeMismatch, match="BACKGROUND_MODE"): cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide", background="off") def test_scope_sees_the_background_axis(self, bg): """★계획은 on 으로 짓고 실행은 off 로 가면 선다.""" bg("on") sc_on = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") built = {"scenario": dict(sc_on), "dimensions": cr.fixture_dimensions("period_episode")} bg("off") sc_off = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") with pytest.raises(cr.ScopeMismatch, match="background"): cr.assert_scope(built, sc_off) class TestFlippingTheSettingOnAFinishedRun: def test_the_folding_scan_reads_the_registry(self): got = cr.steps_folding_setting("background_mode") assert "floor_plan_render" in got and "background_prompt" in got for sid in ("shot_director", "reference_acquisition", "scene_detail", "shot_selection", "grounding_chunk"): assert sid not in got, sid def _run_dir(self, tmp_path, monkeypatch, *, completed, prev_output): root = tmp_path / "run" ep = root / "projects" / "p1" / "checkpoints" / "episodes" / "e1" for sid in completed: (ep / sid).mkdir(parents=True, exist_ok=True) (ep / sid / "manifest.json").write_text( json.dumps({"status": "completed"}), encoding="utf-8") out = root / "canary_run.json" root.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(prev_output), encoding="utf-8") monkeypatch.setattr(cr.ci, "root_dir", lambda _rid: root) monkeypatch.setattr(cr, "run_outputs", lambda _rid: [out]) return root def test_a_flip_with_no_folding_cp_is_safe(self, tmp_path, monkeypatch, bg): self._run_dir(tmp_path, monkeypatch, completed=["shot_director", "reference_acquisition"], prev_output={"scenario": {"background": "on"}}) bg("off") sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") got = cr.assert_setting_flip_is_safe("r", sc) assert got["flipped"] is True and got["stale"] == [] def test_a_flip_over_a_folding_cp_is_refused(self, tmp_path, monkeypatch, bg): """★양성 대조 — floor_plan_render CP 가 있으면 같은 run 재개가 아니다.""" self._run_dir(tmp_path, monkeypatch, completed=["shot_director", "floor_plan_render"], prev_output={"scenario": {"background": "on"}}) bg("off") sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") with pytest.raises(cr.ScopeMismatch, match="floor_plan_render"): cr.assert_setting_flip_is_safe("r", sc) def test_no_flip_passes_even_over_a_folding_cp(self, tmp_path, monkeypatch, bg): self._run_dir(tmp_path, monkeypatch, completed=["floor_plan_render"], prev_output={"scenario": {"background": "on"}}) bg("on") sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") assert cr.assert_setting_flip_is_safe("r", sc)["flipped"] is False def test_an_unknown_previous_counts_as_flipped(self, tmp_path, monkeypatch, bg): """★앞 판 모드를 모르면 뒤집힌 것으로 본다 — 「모름」이 통과가 되면 안 된다.""" self._run_dir(tmp_path, monkeypatch, completed=["floor_plan_render"], prev_output={"scenario": {}}) bg("on") sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") with pytest.raises(cr.ScopeMismatch): cr.assert_setting_flip_is_safe("r", sc) def test_the_real_run_can_flip(self): """실제 run 69e821758f3d — 끝난 CP 중 배경 설정을 접는 것이 없다.""" done = cr.completed_steps_of("69e821758f3d") if not done: pytest.skip("canary 산출이 이 기계에 없다") assert not set(done) & set(cr.steps_folding_setting("background_mode")) class TestTheOutboundDoorsAreZeroOnceTheCentralStepIsDone: """★Codex BLOCK 2026-09-02 — attempt 마다 검색·받기 문이 used 0 으로 새로 생기므로, 중앙 스텝이 끝난 재개 판은 문 자체를 0 으로 열어야 한다.""" def _run_dir(self, tmp_path, monkeypatch, *, completed): root = tmp_path / "run" ep = root / "projects" / "p1" / "checkpoints" / "episodes" / "e1" for sid, st in completed.items(): (ep / sid).mkdir(parents=True, exist_ok=True) (ep / sid / "manifest.json").write_text( json.dumps({"status": st}), encoding="utf-8") monkeypatch.setattr(cr.ci, "root_dir", lambda _rid: root) def test_a_fresh_run_keeps_the_table(self, tmp_path, monkeypatch): self._run_dir(tmp_path, monkeypatch, completed={}) got = cr.outbound_doors_on_resume("r", cr.approved_for("v2_chunk")) assert got["search"] == 120 and got["download"] == 240 def test_a_partial_central_step_keeps_the_doors_open(self, tmp_path, monkeypatch): """★partial 은 다시 돈다 — 문을 닫으면 되볼 것을 못 산다.""" self._run_dir(tmp_path, monkeypatch, completed={cr.CENTRAL_STEP: "partial"}) got = cr.outbound_doors_on_resume("r", cr.approved_for("v2_chunk")) assert got["search"] == 120 and got["download"] == 240 def test_a_completed_central_step_closes_both_doors(self, tmp_path, monkeypatch): """★2026-09-03 뒤집음(Codex 선택 A): 문은 중앙 스텝 하나가 아니라 **검색·받기 소비자 계약** (`OUTBOUND_CONSUMER_STEPS` — 중앙 + 야외 형태 참조) 전부가 끝나야 0 이다. 중앙만 끝난 판은 `test_outbound_doors_follow_the_consumer_contract` 가 「열림」으로 잠근다.""" from app.modules.pipeline.grounding_outbound_consumers import OUTBOUND_CONSUMER_STEPS self._run_dir(tmp_path, monkeypatch, completed={**dict.fromkeys(OUTBOUND_CONSUMER_STEPS, "completed"), "shot_director": "completed"}) got = cr.outbound_doors_on_resume("r", cr.approved_for("v2_chunk")) assert got["search"] == 0 and got["download"] == 0 assert got["counted"] == 480, "★글 상한은 그대로다" def test_the_zeroed_door_really_refuses_at_provider_zero(self): """★양성 대조 — cap 0 문은 첫 예약부터 선다 · cap 1 은 한 번만 지난다.""" from tools.grounding_audit import canary_outbound_gates as og shut = og.OutboundBudget("검색 요청", 0) with pytest.raises(Exception): shut.reserve(where="시험") one = og.OutboundBudget("검색 요청", 1) one.reserve(where="시험") with pytest.raises(Exception): one.reserve(where="시험") def test_the_real_run_closes_the_doors(self): done = cr.completed_steps_of("69e821758f3d", statuses=("completed",)) if not done: pytest.skip("canary 산출이 이 기계에 없다") got = cr.outbound_doors_on_resume("69e821758f3d", cr.approved_for("v2_chunk")) assert (got["search"], got["download"]) == (0, 0) class TestAReopenedCentralStepKeepsTheDoors: """★실측 2026-09-03 새벽(attempt d75a52a0): 다시 연 조사 스텝이 「검색 요청 승인 0」에 막혀 전부 빈손.""" def test_reopen_keeps_the_table_even_when_completed(self, tmp_path, monkeypatch): from tools.grounding_audit import canary_run as cr root = tmp_path ep = root / "projects" / "p" / "checkpoints" / "episodes" / "e" from app.modules.pipeline.grounding_outbound_consumers import OUTBOUND_CONSUMER_STEPS # ★2026-09-03 뒤집음: 소비자 계약의 스텝이 **전부** 끝나야 닫힌 문이다(중앙 하나가 아니다). for sid, st in {**dict.fromkeys(OUTBOUND_CONSUMER_STEPS, "completed"), "shot_director": "completed"}.items(): (ep / sid).mkdir(parents=True, exist_ok=True) (ep / sid / "manifest.json").write_text(__import__("json").dumps({"status": st}), encoding="utf-8") monkeypatch.setattr(cr.ci, "root_dir", lambda _rid: root) closed = cr.outbound_doors_on_resume("r", cr.approved_for("v2_chunk")) assert closed["search"] == 0 and closed["download"] == 0 opened = cr.outbound_doors_on_resume("r", cr.approved_for("v2_chunk"), reopen=(cr.CENTRAL_STEP,)) assert opened["search"] == 120 and opened["download"] == 240, "★다시 열었다는 것이 재승인이다" def test_steps_scope_keeps_the_doors_only_when_the_central_step_is_named(self): from tools.grounding_audit import canary_run as cr a = cr.approved_for("v2_chunk") assert cr.steps_scope(a, ["shot_director"])["approved"]["search"] == 0 got = cr.steps_scope(a, [cr.CENTRAL_STEP])["approved"] assert got["search"] == a["search"] and got["download"] == a["download"]