"""fixture 를 스텝별로 돌리는 주행기. ★유료 0 · 아무것도 안 산다. Codex (2026-08-31) — 예산은 주행 내내 팔을 든 채로 두고, 뜻밖의 전송이 생기면 **per-step delta** 에 남게 한다. 무료로 분류한 스텝이 보내면 그 자리에서 선다. """ from __future__ import annotations import json import pathlib import pytest from tools.grounding_audit import canary_cost_table as ct from tools.grounding_audit import canary_pipeline as cp # ★주행기와 **같은 한 벌**을 쓴다 — 사본을 두면 계획이 갈린다 from tools.grounding_audit.canary_run import default_fixture_config # noqa: E402 CFG = default_fixture_config() @pytest.fixture def plan(): return ct.execution_plan(config=CFG) @pytest.fixture def root(monkeypatch, tmp_path): monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) # ★★대역도 **체크포인트를 남긴다** — production 이 그러기 때문이다. # 안 남기면 `assert_step_finished` 가 「안 돌았다」로 읽는다(맞는 판정). from app.core.config import settings monkeypatch.setattr(settings, "projects_dir", str(tmp_path / "projects")) return tmp_path def _finish(kw, step, status="completed"): """대역이 그 스텝을 **끝냈다고 파일에 남긴다**. ★production 과 같은 자리.""" import json as _json from app.core.config import settings d = (pathlib.Path(settings.projects_dir) / kw["project_id"] / "checkpoints" / "episodes" / kw["episode_id"] / step) d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text(_json.dumps({"status": status}), encoding="utf-8") class TestItDoesNotRunWithoutLive: def test_a_dry_call_runs_nothing(self, plan, root): got = cp.run_pipeline(run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps={s: 1 for s in plan["applied_metered"]}, emergency_counted=92, approved_image_calls=0) assert got["live"] is False and got["per_step"] == {} assert "안 돌렸다" in got["note"] def test_the_free_steps_are_named_up_front(self, plan, root): got = cp.run_pipeline(run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps={}, emergency_counted=92, approved_image_calls=0) assert set(got["free_steps"]) == set(plan["applied_free"]) class TestOneStepAtATime: def _run(self, plan, root, *, sends, caps=None, ceiling=92, before_step=None): """`run_steps_batch` 를 갈아 끼워 **보내는 시늉**만 한다.""" from app.core.research_call_budget import reserve_current_research_call from app.services import analysis_dispatch_service as ads called = [] def fake(**kw): s = kw["step_ids"][0] called.append(s) _finish(kw, s) for _ in range(sends.get(s, 0)): reserve_current_research_call(source=f"fake[{s}]") import unittest.mock as m with m.patch.object(ads, "run_steps_batch", fake): got = cp.run_pipeline( run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps=caps if caps is not None else {s: 99 for s in plan["applied"] and [r["step"] for r in plan["applied"]]}, emergency_counted=ceiling, approved_image_calls=0, live=True, before_step=before_step) return got, called def test_a_gate_that_recomputes_a_larger_cap_opens_the_step_that_wide(self, plan, root): """★새 run 의 중앙 상한이 선언 하한으로 작게 잡혔을 때 — 문이 실제 의무 수로 더 큰 상한을 내면 그 수로 연다(서지 않는다). 남은 ceiling(이 harness 92) 안이어야 한다.""" paid = [r["step"] for r in plan["applied"]] target = paid[0] caps = {s: (1 if s == target else 99) for s in paid} got, called = self._run(plan, root, sends={}, caps=caps, before_step=lambda s: ( {"checked": True, "recomputed_logical_cap": 5, "why": "계획 1 < 실제 5"} if s == target else None)) assert target in called rec = got["per_step"][target].get("step_cap_recomputed") assert rec and rec["to"] > rec["from"] and rec["to"] == 5 * cp._per_logical(plan) assert rec["to"] <= rec["remaining_ceiling"] def test_a_recompute_beyond_the_remaining_ceiling_stops_before_the_step(self, plan, root): """★재산정한 상한이 run 전체의 남은 ceiling 을 넘으면 그때만 선다 (Codex 06:10) — 정지선은 우회 불가.""" paid = [r["step"] for r in plan["applied"]] target = paid[0] with pytest.raises(cp.CanaryStopped) as e: self._run(plan, root, sends={}, before_step=lambda s: ( {"checked": True, "recomputed_logical_cap": 1000, "why": "계획 5 < 실제 1000"} if s == target else None)) assert "남은 ceiling" in str(e.value) def test_a_producer_gate_runs_before_the_step_and_a_raise_stops_before_it(self, plan, root): """★producer 상한 문(Codex BLOCK 2026-09-03) — 살 스텝을 **부르기 직전**에 불리고, 던지면 `run_steps_batch` 는 그 스텝에 **안 닿는다**. 반환값은 per_step 에 남는다.""" paid = [r["step"] for r in plan["applied"]] target = paid[1] order = [] def gate(s): order.append(("gate", s)) if s == target: raise RuntimeError("실외 그룹 3 > 상한 2") return {"checked": True} with pytest.raises(RuntimeError): got, called = self._run(plan, root, sends={}, before_step=gate) # ★문은 스텝마다 그 스텝 **앞**에서 불렸고, 던진 스텝은 안 불렸다 assert ("gate", target) in order assert order.index(("gate", paid[0])) < order.index(("gate", target)) def test_every_step_of_the_closure_runs_once(self, plan, root): """★★★적용 안 되는 스텝도 **돌아야** 한다 (실측 2026-09-01). `check_gate` 는 의존 스텝의 **`step_run` 기록**을 본다 — 기록이 아예 없으면 막는다. 「적용 안 됨」은 **돌아서 그렇게 찍혀야** 뒤가 지나간다. 앞 판은 `grounding_a0` 를 목록에서 빼서 `entity_all_character` 가 막혔다(정적으로는 `not_applicable` 인데도). """ got, called = self._run(plan, root, sends={}) closure = ct.execution_steps_to("scene_detail") assert called == closure assert len(got["per_step"]) == len(closure) # ★건너뛴다고 적었던 것도 **목록에 있다** assert "grounding_a0" in called def test_a_step_outside_the_paid_list_may_not_send(self, plan, root): """★유료 목록 밖 스텝이 보내면 **선다**. ★★2026-09-01 에 **더 세졌다** (Codex BLOCK). 앞에는 스텝이 **끝난 뒤** delta 를 보고 `CanaryStopped` 로 섰다 — 그건 문이 아니라 사후 경보기라, provider 는 이미 불린 뒤였다. 이제 **문 앞에서** 거절된다. 여기서 잠그는 것은 「선다」가 아니라 **「provider 에 안 닿는다」**다. """ from app.core.research_call_budget import ResearchCallBudgetExceeded outside = [r["step"] for r in plan["skipped"]][0] with pytest.raises(ResearchCallBudgetExceeded) as e: self._run(plan, root, sends={outside: 1}) assert "상한" in str(e.value) def test_a_free_step_that_sends_stops_the_run(self, plan, root): """★★무료로 적은 스텝이 보내면 **문 앞에서** 거절된다.""" from app.core.research_call_budget import ResearchCallBudgetExceeded free = plan["applied_free"][0] with pytest.raises(ResearchCallBudgetExceeded) as e: self._run(plan, root, sends={free: 1}) assert free in str(e.value) def test_going_over_a_step_cap_stops(self, plan, root): """★그 스텝 몫을 넘기면 **넘긴 그 호출이** 거절된다.""" from app.core.research_call_budget import ResearchCallBudgetExceeded metered = plan["applied_metered"][0] with pytest.raises(ResearchCallBudgetExceeded): self._run(plan, root, sends={metered: 50}, caps={s: 1 for s in [r["step"] for r in plan["applied"]]}) def test_the_ceiling_ends_the_run_as_inconclusive(self, plan, root): metered = plan["applied_metered"][0] with pytest.raises(cp.CanaryStopped) as e: self._run(plan, root, sends={metered: 4}, ceiling=4, caps={s: 99 for s in [r["step"] for r in plan["applied"]]}) assert "inconclusive" in str(e.value) assert "자동으로 늘리거나 다시" in str(e.value) def test_it_writes_after_every_step(self, plan, root): """★한 스텝마다 내려쓴다 — 죽어도 어디까지 갔는지 남는다.""" self._run(plan, root, sends={}) p = root / "canary_a1b2c3d4" / "pipeline_run.json" got = json.loads(p.read_text(encoding="utf-8")) assert len(got["per_step"]) == len(ct.execution_steps_to("scene_detail")) def test_a_crash_records_where_it_stopped(self, plan, root): from app.services import analysis_dispatch_service as ads def boom(**kw): if kw["step_ids"][0] == plan["applied"][1]["step"]: raise RuntimeError("일부러") _finish(kw, kw["step_ids"][0]) import unittest.mock as m with m.patch.object(ads, "run_steps_batch", boom): with pytest.raises(RuntimeError): cp.run_pipeline(run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps={s: 99 for s in [r["step"] for r in plan["applied"]]}, emergency_counted=92, approved_image_calls=0, live=True) got = json.loads((root / "canary_a1b2c3d4" / "pipeline_run.json") .read_text(encoding="utf-8")) assert got["stopped_at"]["step"] == plan["applied"][1]["step"] class TestTheCeilingIsForTheWholeRunNotOneAttempt: """★★★재개마다 상한이 **처음부터 다시** 열리던 것 (Codex BLOCK 2026-09-01). 같은 run 을 여러 번 재개하면 매번 92 를 새로 써서 **누계가 92 를 훌쩍 넘길 수 있었다.** 그리고 `pipeline_run.json` 이 덮여 첫 attempt 의 per-step 기록이 **사라졌다**. """ def _run(self, plan, root, *, sends, ceiling=92, finish=None): from app.core.research_call_budget import reserve_current_research_call from app.services import analysis_dispatch_service as ads def fake(**kw): s = kw["step_ids"][0] # ★★★**이미 끝난 스텝은 provider 를 안 부른다** — 진짜 스텝도 # 체크포인트에서 돌아온다. 대역이 그래도 부르면 문에 걸려 # 거절당하는데, 그것은 대역이 production 과 다른 것이다. already = cp.durable_status(kw["project_id"], kw["episode_id"], s) in cp.REUSE_ZERO_STATES if finish is None or s in finish: _finish(kw, s) if already: return for _ in range(sends.get(s, 0)): reserve_current_research_call(source=f"fake[{s}]") import unittest.mock as m with m.patch.object(ads, "run_steps_batch", fake): return cp.run_pipeline( run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps={s: 99 for s in plan["applied_metered"]}, emergency_counted=ceiling, approved_image_calls=0, live=True) def test_two_attempts_both_survive(self, plan, root): """★★덧붙이기만 한다 — 앞 것을 **안 덮는다**. ★2026-09-02: 재개는 **아직 안 끝난 스텝**을 돈다. 앞 판에 끝난 스텝은 이제 provider 앞에서 `cap=0` 이라 다시 못 산다(Codex BLOCK). """ m = plan["applied_metered"][0] self._run(plan, root, sends={m: 3}) self._run(plan, root, sends={m: 2}) got = cp.read_attempts(root / "canary_a1b2c3d4") assert len(got) == 2, f"★덮었다: {len(got)}" # ★★둘째는 **0** 이다 — 앞 판에 끝난 스텝은 provider 앞에서 `cap=0` # 이라 다시 못 산다 (2026-09-02 Codex BLOCK). 앞 판은 3 그대로다. assert [a["used"] for a in got] == [3, 0] assert got[0]["attempt_id"] != got[1]["attempt_id"] def test_the_third_attempt_sees_the_cumulative(self, plan, root): """★셋째는 **남은 것**만 연다 — 92 를 새로 안 연다.""" m = plan["applied_metered"][0] self._run(plan, root, sends={m: 3}) self._run(plan, root, sends={m: 2}) # ★0 이다 — 이미 끝났다 third = self._run(plan, root, sends={}) assert third["cumulative_before"] == 3 assert third["scope_cap"] == 92 - 3 def test_it_stops_at_the_run_wide_ceiling(self, plan, root): """★★★누계가 정지선에 닿으면 **선다**. ★2026-09-02: 「attempt 별로 다시 안 연다」는 이제 아래 `test_nothing_left_refuses_before_provider` 가 잰다 — 끝난 스텝은 재개에서 `cap=0` 이라 둘째 판이 애초에 못 산다. """ m = plan["applied_metered"][0] with pytest.raises(cp.CanaryStopped) as e: self._run(plan, root, sends={m: 6}, ceiling=6) assert "누적이 정지선" in str(e.value) def test_nothing_left_refuses_before_provider(self, plan, root): """★★남은 것이 0 이면 **아무것도 안 사고** 선다. ★첫 판은 정지선에 닿아 스스로 서고, 그 사용량은 장부에 남는다. 그 다음 판은 **열지도 못한다** — 이것이 재개 반복을 막는 자리다. """ m = plan["applied_metered"][0] with pytest.raises(cp.CanaryStopped): self._run(plan, root, sends={m: 5}, ceiling=5) assert cp.cumulative_used(root / "canary_a1b2c3d4") == 5 with pytest.raises(cp.LedgerRefused) as e: self._run(plan, root, sends={m: 1}, ceiling=5) assert "남은 것이" in str(e.value) def test_a_broken_ledger_stops(self, root): p = root / "canary_a1b2c3d4" p.mkdir(parents=True, exist_ok=True) (p / cp.ATTEMPTS).write_text("{망가진", encoding="utf-8") with pytest.raises(cp.LedgerRefused): cp.cumulative_used(p) def test_an_attempt_without_a_number_stops(self, root): """★「모르는」 attempt 를 **0 으로 발명하지 않는다**.""" p = root / "canary_a1b2c3d4" p.mkdir(parents=True, exist_ok=True) (p / cp.ATTEMPTS).write_text( json.dumps([{"attempt_id": "x", "used": None}]), encoding="utf-8") with pytest.raises(cp.LedgerRefused) as e: cp.cumulative_used(p) assert "발명하지 않는다" in str(e.value) def test_a_stop_reason_is_recorded_in_the_attempt(self, plan, root): """★목록 밖 스텝이 **문 앞에서 거절된** 사유도 attempt 에 남는다. ★사유 글이 바뀌었다 — 앞에는 「유료 목록에 없는데 N번 보냈다」였다. 이제는 보내기 **전에** 막히므로 「이 스텝의 상한」이 사유다. """ from app.core.research_call_budget import ResearchCallBudgetExceeded outside = [r["step"] for r in plan["skipped"]][0] with pytest.raises(ResearchCallBudgetExceeded): self._run(plan, root, sends={outside: 1}) got = cp.read_attempts(root / "canary_a1b2c3d4") assert got[-1]["status"] == "stopped" why = (got[-1]["stopped_at"] or {})["why"] assert outside in why and "상한" in why # ★막은 것이 장부에 남아야 「안 썼다」와 갈린다 assert got[-1]["per_step"][outside]["denied_by_step_cap"] == 1 class TestAKilledProcessStillLeavesATrace: """★★★프로세스가 **통째로 사라져도** 장부에 남아야 한다. Codex BLOCK (2026-09-01) — 앞 판은 attempt 를 **끝나고 나서** 적었다. 그래서 provider 를 부른 뒤 `SIGKILL`·`os._exit` 이 오면 **한 줄도 안 남고**, 다음 재개가 그만큼을 **덜 세어** 상한을 넘길 수 있었다. ★일반 `RuntimeError` 시험으로 대신할 수 없다 — 그건 파이썬이 잡아서 닫아 주는 판이다. **진짜로 죽여야** 잰다. """ CHILD = ''' import os, sys, json sys.path.insert(0, {backend!r}) sys.path.insert(0, {tests!r}) os.environ["THEROAD_CANARY_ROOT"] = {rootenv!r} from app.core.research_call_budget import reserve_current_research_call from app.services import analysis_dispatch_service as ads from tools.grounding_audit import canary_cost_table as ct from tools.grounding_audit import canary_pipeline as cp from tools.grounding_audit.canary_run import default_fixture_config FIXTURE_CONFIG = default_fixture_config() plan = ct.execution_plan(config=FIXTURE_CONFIG) # ★사본을 안 만든다 plan["target"] = "scene_detail" def fake(**kw): # ★한 번 사고 **그 자리에서 죽는다** — 파이썬이 못 잡는다 reserve_current_research_call(source="child") os._exit(9) ads.run_steps_batch = fake cp.run_pipeline(run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps={{s: 99 for s in plan["applied_metered"]}}, emergency_counted=92, approved_image_calls=0, live=True) ''' def _backend(self): """★`cwd` 가 아니라 **모듈 자리**에서 뽑는다. 저장소 뿌리에서 pytest 를 돌리면 자식이 `app` 을 못 찾아 `os._exit` 까지 못 가고, 그러면 이 시험이 **계약이 아니라 경로**를 잰다 (Codex 2026-09-01). """ from pathlib import Path return Path(cp.__file__).resolve().parents[2] def test_the_open_attempt_survives_the_kill(self, root): import subprocess import sys as _s backend = self._backend() code = self.CHILD.format(backend=str(backend), tests=str(backend / "tests"), rootenv=str(root)) got = subprocess.run([_s.executable, "-c", code], capture_output=True, text=True, cwd=str(backend)) # ★`.env` 도 여기서 읽힌다 assert got.returncode == 9, got.stderr[-500:] # ★부모가 되읽으면 **열린 판이 남아 있다** p = root / "canary_a1b2c3d4" live = cp.open_attempts(p) assert len(live) == 1, f"★죽은 판이 장부에 없다: {cp.read_attempts(p)}" assert live[0]["status"] == "running" assert live[0]["used"] is None, "★얼마 썼는지 아는 척하면 안 된다" def test_the_next_run_buys_nothing(self, plan, root): """★★★그 뒤 판은 **provider 를 한 번도 안 부른다**.""" import subprocess import sys as _s backend = self._backend() code = self.CHILD.format(backend=str(backend), tests=str(backend / "tests"), rootenv=str(root)) killed = subprocess.run([_s.executable, "-c", code], capture_output=True, text=True, cwd=str(backend)) assert killed.returncode == 9, killed.stderr[-400:] calls = [] def fake(**kw): calls.append(kw["step_ids"][0]) _finish(kw, kw["step_ids"][0]) import unittest.mock as m from app.services import analysis_dispatch_service as ads with m.patch.object(ads, "run_steps_batch", fake): with pytest.raises(cp.LedgerRefused) as e: cp.run_pipeline(run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps={s: 99 for s in plan["applied_metered"]}, emergency_counted=92, approved_image_calls=0, live=True) assert "아직 열려 있다" in str(e.value) assert calls == [], f"★열린 판이 있는데 돌았다: {calls}" def test_two_attempts_cannot_be_open_at_once(self, root): p = root / "canary_a1b2c3d4" cp.reserve_attempt(p, {"attempt_id": "하나"}) with pytest.raises(cp.LedgerRefused) as e: cp.reserve_attempt(p, {"attempt_id": "둘"}) assert "안 닫힌 attempt" in str(e.value) def test_an_open_attempt_is_not_counted_as_zero(self, root): p = root / "canary_a1b2c3d4" cp.reserve_attempt(p, {"attempt_id": "하나"}) with pytest.raises(cp.LedgerRefused) as e: cp.cumulative_used(p) assert "0 으로 세지 않는다" in str(e.value) class TestTheTextLedgerCannotSeeImages: """★★★글 예산은 이미지 문을 **못 본다** (실측 2026-09-01). `scene_detail` 까지의 closure 안에 `floor_plan_render` 가 있고 그것은 `gpt-image-2` 를 산다. 앞 판 그대로 돌렸으면 글 장부에는 **0 으로 적히고** 실제로는 돈이 나갔다. """ def _run(self, plan, root, *, buys, approved=0): from app.core.image_call_budget import reserve_current_call from app.services import analysis_dispatch_service as ads def fake(**kw): s = kw["step_ids"][0] _finish(kw, s) for _ in range(buys.get(s, 0)): reserve_current_call(source=f"fake[{s}]") import unittest.mock as m with m.patch.object(ads, "run_steps_batch", fake): return cp.run_pipeline( run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps={s: 99 for s in plan["applied_metered"]}, emergency_counted=92, approved_image_calls=approved, live=True) def test_an_unapproved_image_stops_at_the_door(self, plan, root): """★승인 0 이면 **문 앞에서** 선다 — 사고 나서 세는 것이 아니다.""" from app.core.image_call_budget import ImageCallBudgetExceeded with pytest.raises(ImageCallBudgetExceeded): self._run(plan, root, buys={"floor_plan_render": 1}) got = json.loads((root / "canary_a1b2c3d4" / "pipeline_run.json") .read_text(encoding="utf-8")) assert got["stopped_at"]["step"] == "floor_plan_render" assert got["image_budget"]["used"] == 0 assert got["image_budget"]["denied"] == 1 # ★죽은 스텝의 몫이 장부에 있어야 「안 썼다」로 안 읽힌다 assert got["per_step"]["floor_plan_render"]["image_denied"] == 1 def test_the_ledger_writes_images_apart_from_text(self, plan, root): """★합치면 **무엇에 돈이 나갔는지** 못 읽는다.""" from app.core.image_call_budget import ImageCallBudgetExceeded with pytest.raises(ImageCallBudgetExceeded): self._run(plan, root, buys={"floor_plan_render": 1}) led = json.loads((root / "canary_a1b2c3d4" / "pipeline_attempts.json") .read_text(encoding="utf-8")) assert led[-1]["image_denied"] == 1 assert led[-1]["image_used"] == 0 assert led[-1]["used"] == 0 # ★글은 한 번도 안 나갔다 assert led[-1]["approved_image_calls"] == 0 def test_a_step_that_buys_within_approval_goes_on(self, plan, root): """★★뒤집은 시험 (2026-09-02 stage2a 실측): 승인 80 인데 옛 규칙이 승인값을 안 보고 ref_image_gen 4장 뒤에 세웠다. 승인이 있으면 정지선은 run 전체 이미지 문이다 — 승인 안에서 산 것으로는 서지 않는다.""" got = self._run(plan, root, buys={"floor_plan_render": 1}, approved=3) assert got["stopped_at"] is None assert got["image_budget"]["used"] == 1 assert got["per_step"]["floor_plan_render"]["image_counted"] == 1 led = json.loads((root / "canary_a1b2c3d4" / "pipeline_attempts.json") .read_text(encoding="utf-8")) assert led[-1]["image_used"] == 1 and led[-1]["image_scope_cap"] == 3 def test_beyond_the_run_wide_image_cap_the_door_stops(self, plan, root): """★승인 안이라도 문(run 전체 남은 것)을 넘으면 provider 앞에서 선다.""" from app.core.image_call_budget import ImageCallBudgetExceeded with pytest.raises(ImageCallBudgetExceeded): self._run(plan, root, buys={"floor_plan_render": 5}, approved=3) saved = json.loads((root / "canary_a1b2c3d4" / "pipeline_run.json") .read_text(encoding="utf-8")) assert saved["image_budget"]["used"] == 3 assert saved["image_budget"]["denied"] == 1 def test_a_free_step_that_touches_the_image_door_is_a_misclassification( self, plan, root): free = plan["applied_free"][0] with pytest.raises(Exception): self._run(plan, root, buys={free: 1}, approved=1) saved = json.loads((root / "canary_a1b2c3d4" / "pipeline_run.json") .read_text(encoding="utf-8")) assert saved["stopped_at"]["step"] == free def test_nothing_touching_images_leaves_zeros(self, plan, root): got = self._run(plan, root, buys={}) assert got["image_budget"] == {"cap": 0, "used": 0, "denied": 0, "remaining": 0} assert all(v["image_counted"] == 0 for v in got["per_step"].values()) class TestTheGateStandsBeforeTheProviderNotAfter: """★★★Codex BLOCK (2026-09-01) — 앞 판은 **사후 경보기**였다. 스텝이 끝난 뒤 delta 를 보고 섰으므로, 목록 밖 스텝 (`visual_continuity_anchor`)이 **provider 를 부르고 나서야** 잡혔다. 분류를 고쳐도 다음 실수 때 또 승인 밖 호출이 나간다. 그래서 **부르기 전에** 그 스텝 몫만 문에 건다. """ def _run(self, plan, root, *, sends, caps): """★provider 자리에서 **실제로 나간 수**를 따로 센다.""" from app.core.research_call_budget import ( ResearchCallBudgetExceeded, reserve_current_research_call) from app.services import analysis_dispatch_service as ads outbound = [] def fake(**kw): s = kw["step_ids"][0] _finish(kw, s) for _ in range(sends.get(s, 0)): # ★문을 지난 것만 provider 로 간다 reserve_current_research_call(source=f"fake[{s}]") outbound.append(s) import unittest.mock as m with m.patch.object(ads, "run_steps_batch", fake): try: got = cp.run_pipeline( run_id="a1b2c3d4", project_id="p", episode_id="e", plan=plan, caps=caps, emergency_counted=92, approved_image_calls=0, live=True) except (ResearchCallBudgetExceeded, cp.CanaryStopped) as exc: got = {"raised": exc} return got, outbound def test_a_step_outside_the_paid_list_never_reaches_the_provider( self, plan, root): """★이번에 실제로 난 것 — 목록 밖 스텝이 1번 보냈다.""" outside = [r["step"] for r in plan["skipped"]][0] got, outbound = self._run( plan, root, sends={outside: 1}, caps={s: 99 for s in plan["applied_metered"]}) assert outbound == [], f"★승인 밖인데 provider 로 {outbound} 나갔다" saved = json.loads((root / "canary_a1b2c3d4" / "pipeline_run.json") .read_text(encoding="utf-8")) row = saved["per_step"][outside] assert row["step_cap"] == 0 assert row["denied_by_step_cap"] == 1 # ★막은 것이 기록된다 assert row["counted"] == 0 # ★run 예산은 안 늘었다 assert saved["budget"]["used"] == 0 def test_a_free_step_never_reaches_the_provider_either(self, plan, root): free = plan["applied_free"][0] got, outbound = self._run( plan, root, sends={free: 1}, caps={s: 99 for s in plan["applied_metered"]}) assert outbound == [] saved = json.loads((root / "canary_a1b2c3d4" / "pipeline_run.json") .read_text(encoding="utf-8")) assert saved["per_step"][free]["denied_by_step_cap"] == 1 def test_a_metered_step_beyond_its_own_cap_is_cut_at_its_cap(self, plan, root): """★★run 상한이 남아 있어도 **그 스텝 몫**에서 끊긴다.""" first = plan["applied_metered"][0] got, outbound = self._run(plan, root, sends={first: 5}, caps={first: 2}) # ★per_logical 이 1 이면 2 번만 나간다 per = cp._per_logical(plan) assert len(outbound) == 2 * per, outbound saved = json.loads((root / "canary_a1b2c3d4" / "pipeline_run.json") .read_text(encoding="utf-8")) assert saved["per_step"][first]["denied_by_step_cap"] >= 1 def test_the_run_wide_cap_still_holds_at_the_same_time(self, plan, root): """★둘 다 살아 있어야 한다 — 좁은 문을 넓혀도 run 정지선은 그대로.""" metered = plan["applied_metered"] got, outbound = self._run(plan, root, sends={s: 99 for s in metered}, caps={s: 99 for s in metered}) assert len(outbound) <= 92 class TestAReopenPassRunsOnlyWhatItReopened: """★★★closure 전부를 돌면 `partial` 스텝이 **제 실패를 다시 시도**한다 (2026-09-02 실측). 그런데 그 스텝은 `already` 라 상한이 0 이라 전부 거절되고, 그 거절이 문을 세운다. `entity_t2i` 의 거절 16건이 남은 상한 23을 다 먹어 정작 하려던 재판정이 **시작도 못 했다**. 재판정은 앞 스텝을 하나도 안 쓴다 — 체크포인트가 이미 다 있다. """ def test_it_narrows_the_step_list(self, monkeypatch): from tools.grounding_audit import canary_pipeline as cp from tools.grounding_audit import canary_cost_table as ct monkeypatch.setattr(ct, "execution_steps_to", lambda _t: ["a", "b", "reference_acquisition"]) got = cp.run_pipeline( run_id="deadbeef", project_id="p", episode_id="e", plan={"target": "reference_acquisition", "applied_free": [], "applied_metered": []}, caps={}, emergency_counted=1, approved_image_calls=0, reopen=("reference_acquisition",), live=False) assert got["steps"] == ["reference_acquisition"] assert got["reopened"] == ["reference_acquisition"] def test_a_normal_pass_keeps_every_step(self, monkeypatch): from tools.grounding_audit import canary_pipeline as cp from tools.grounding_audit import canary_cost_table as ct monkeypatch.setattr(ct, "execution_steps_to", lambda _t: ["a", "b", "reference_acquisition"]) got = cp.run_pipeline( run_id="deadbeef", project_id="p", episode_id="e", plan={"target": "reference_acquisition", "applied_free": [], "applied_metered": []}, caps={}, emergency_counted=1, approved_image_calls=0, live=False) assert got["steps"] == ["a", "b", "reference_acquisition"] assert got["reopened"] == [] def test_an_unknown_step_stops(self, monkeypatch): """★모르는 이름이면 **무엇을 도는지 모른다** — 안 연다.""" import pytest from tools.grounding_audit import canary_pipeline as cp from tools.grounding_audit import canary_cost_table as ct monkeypatch.setattr(ct, "execution_steps_to", lambda _t: ["a"]) with pytest.raises(cp.CanaryStopped, match="closure 에 없다"): cp.run_pipeline( run_id="deadbeef", project_id="p", episode_id="e", plan={"target": "a", "applied_free": [], "applied_metered": []}, caps={}, emergency_counted=1, approved_image_calls=0, reopen=("nope",), live=False) class TestPartialIsNotACapZeroReuse: """★★Codex 재리뷰 (2026-09-02): `partial` 은 의존 통과 상태일 수는 있어도 **cap 0 되쓰기 상태가 아니다** — production resume 이 그 스텝의 남은 일을 다시 사므로 그 몫의 상한이 열려 있어야 한다. 실측: entity_t2i partial 의 재시도 27건이 cap 0 에서 전부 거절되고 그 거절이 문을 세웠다(유료 0). """ def test_the_reuse_set_excludes_partial(self): from tools.grounding_audit import canary_pipeline as cp assert "partial" not in cp.REUSE_ZERO_STATES assert set(cp.REUSE_ZERO_STATES) == {"completed", "not_applicable"} assert "partial" in cp.FINISHED_STATES, "★의존 통과 상태 목록은 그대로다" def test_the_gate_reads_the_reuse_set_not_the_finished_set(self): """★소스가 cap 0 판정에 `REUSE_ZERO_STATES` 를 쓴다 — 한 벌로 합치지 않는다.""" import inspect from tools.grounding_audit import canary_pipeline as cp # ★cap 0 결정은 `plan_reentry` 한 곳이 한다 (2026-09-02 밤: 지문 어긋남→force 가 더해짐) src = inspect.getsource(cp.plan_reentry) assert "REUSE_ZERO_STATES" in src and "FINISHED_STATES" not in src run_src = inspect.getsource(cp.run_pipeline) i = run_src.index("step_cap = 0") assert 'decision["cap_zero"]' in run_src[max(0, i - 200):i]