"""이미지 상한은 **run 전체**에 걸리고, 이미지 승인은 **정확한 시나리오**에 결속된다. ★Codex 비용 상한 BLOCK 둘 (2026-09-02): ① 글은 `remaining_cap` 으로 누계를 빼고 열었는데 이미지는 attempt 마다 승인값을 통째로 새로 열었다 — 20장 뒤 crash→재개면 40장을 더 열어 승인 40 을 넘는다. ② `APPROVED_IMAGE_CALLS` 는 전역 상수라 0→40 은 「이번 stage2a 승인」이 아니라 다른 canary 범위에도 열린 값이었다. 전이 기록·재개 문·scope 문이 이미지 승인을 안 봤다. """ from __future__ import annotations import json from pathlib import Path import pytest from tools.grounding_audit import canary_pipeline as cp from tools.grounding_audit import canary_run as cr from tools.grounding_audit import canary_isolation as ci @pytest.fixture def root(tmp_path, monkeypatch) -> Path: monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) d = ci.root_dir("4dabc123abc1") d.mkdir(parents=True, exist_ok=True) return d def _attempt(root, aid, *, status="completed", used=0, image_used=None): rec = {"attempt_id": aid, "status": status, "used": used, "started_kst": "x"} if image_used is not None: rec["image_used"] = image_used cp.append_attempt(root, rec) class TestTheImageCapIsCumulative: def test_first_attempt_twenty_then_the_next_opens_only_twenty(self, root): _attempt(root, "a1", used=30, image_used=20) assert cp.cumulative_image_used(root) == 20 assert cp.remaining_image_cap(root, ceiling=40) == 20 def test_a_terminal_attempt_is_not_subtracted_twice(self, root): _attempt(root, "a1", image_used=20) _attempt(root, "a2", image_used=5) assert cp.cumulative_image_used(root) == 25 assert cp.remaining_image_cap(root, ceiling=40) == 15 def test_an_open_attempt_refuses(self, root): _attempt(root, "a1", image_used=20) _attempt(root, "a2", status="running") with pytest.raises(cp.LedgerRefused, match="열려"): cp.cumulative_image_used(root) def test_a_correction_is_the_record(self, root): _attempt(root, "a1", image_used=20) cp.append_event(root, {"kind": cp.EVENT_CORRECTION, "attempt_id": "a1", "image_debit_now": 23, "why": "provider 로그로 3장 더 확인"}) assert cp.cumulative_image_used(root) == 23 def test_old_attempts_without_the_column_count_as_zero(self, root): """이미지 문 전의 판 — 승인 0 으로 돌았고 문이 닫혀 있었다.""" _attempt(root, "a0", used=40) assert cp.cumulative_image_used(root) == 0 def test_ceiling_zero_means_a_closed_door_not_a_refusal(self, root): _attempt(root, "a1", used=10) assert cp.remaining_image_cap(root, ceiling=0) == 0 def test_nothing_left_refuses(self, root): _attempt(root, "a1", image_used=40) with pytest.raises(cp.LedgerRefused, match="남은 것이 없다"): cp.remaining_image_cap(root, ceiling=40) def test_over_the_ceiling_refuses(self, root): _attempt(root, "a1", image_used=41) with pytest.raises(cp.LedgerRefused, match="어긋난다"): cp.remaining_image_cap(root, ceiling=40) def test_the_pipeline_opens_only_the_remainder(self, root, monkeypatch): """★끝점 — `run_pipeline` 이 문을 열 때 넘기는 cap 이 남은 것이다.""" _attempt(root, "a1", image_used=20) opened = {} class _Scope: def __init__(self, cap): self.cap = cap def __enter__(self): return self def __exit__(self, *a): return False def snapshot(self): return {"cap": self.cap, "used": 0, "denied": 0} def _image_scope(cap): opened["image"] = cap # ★이미지 문에 넘긴 cap 만 적는다 return _Scope(cap) from tools.grounding_audit import canary_image_budget as cib monkeypatch.setattr(cib, "canary_image_scope", _image_scope) # ★글 문·outbound 문·run_steps_batch 는 여기서 안 연다 — 이미지 문 인자만 잰다 from tools.grounding_audit import canary_text_budget as ctb monkeypatch.setattr(ctb, "canary_text_scope", lambda cap: _Scope(cap)) from tools.grounding_audit import canary_outbound_gates as cog monkeypatch.setattr(cog, "canary_outbound_scope", lambda search_cap, download_cap: _Scope(0)) monkeypatch.setattr(cog, "snapshot_of", lambda o: {}) plan = {"applied": [], "applied_free": [], "applied_metered": [], "unknown": [], "skipped": [], "target": "x", "contract": {}} got = cp.run_pipeline(run_id="4dabc123abc1", project_id="p", episode_id="e", plan=plan, caps={}, emergency_counted=92, approved_image_calls=40, live=True) assert got["image_cumulative_before"] == 20 assert got["image_scope_cap"] == 20 assert opened["image"] == 20 led = [a for a in cp.read_attempts(root) if a.get("kind") is None] assert led[-1]["image_scope_cap"] == 20 and led[-1]["image_ceiling"] == 40 class TestTheImageApprovalIsBoundToTheExactScope: SC = {"mode": "v2_chunk", "fixture": "period_episode", "target": "scene_image_pipeline", "background": "off", "still_recipe": "off", "outdoor": "off"} def test_unlisted_scope_is_zero(self): # ★이 정확한 범위만 80 이 승인돼 있다(2026-09-02 stage2a) — 다른 target 은 0 assert cr.approved_images_for(self.SC) == 80 assert cr.approved_images_for({**self.SC, "target": "world_guide"}) == 0 def test_only_the_exact_scope_gets_the_value(self, monkeypatch): monkeypatch.setitem(cr.APPROVED_IMAGES_BY_SCOPE, cr.scope_key(self.SC), 40) assert cr.approved_images_for(self.SC) == 40 for axis, other in (("target", "world_guide"), ("fixture", "canary_one_scene"), ("outdoor", "on"), ("still_recipe", "on"), ("background", "on"), ("mode", "legacy")): assert cr.approved_images_for({**self.SC, axis: other}) == 0, axis def test_the_scenario_carries_it(self, monkeypatch): from app.core.config import settings for k, v in (("background_mode", "off"), ("still_recipe_mode", "off"), ("outdoor_lane_plan_enabled", False), ("outdoor_lane_pipe_enabled", False), ("outdoor_direct_compose_enabled", False), ("outdoor_map_conti_enabled", False)): monkeypatch.setattr(settings, k, v) monkeypatch.setitem(cr.APPROVED_IMAGES_BY_SCOPE, cr.scope_key(self.SC), 40) sc = cr.scenario(mode="v2_chunk", fixture="period_episode", target="scene_image_pipeline") assert sc["approved_images"] == 40 assert cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide")["approved_images"] == 0 def test_the_scope_gate_refuses_a_plan_with_another_approval(self, monkeypatch): monkeypatch.setattr(cr, "fixture_dimensions", lambda f: {"d": 1}) built = {"scenario": {**self.SC, "approved": cr.approved_for("v2_chunk"), "approved_images": 40}, "dimensions": {"d": 1}} with pytest.raises(cr.ScopeMismatch, match="이미지 승인"): cr.assert_scope(built, self.SC) built["scenario"]["approved_images"] = cr.approved_images_for(self.SC) assert cr.assert_scope(built, self.SC)["scope"] == "일치" def test_the_resume_gate_compares_it(self, tmp_path, monkeypatch): """전이 줄에 이미지 승인이 없거나 다르면 안 잇는다.""" monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) rid = "5dabc123abc1" d = tmp_path / f"canary_{rid}" d.mkdir(parents=True, exist_ok=True) (d / "canary_run.json").write_text(json.dumps( {"code": {"tip": "a" * 40, "clean": True}, "scenario": {"mode": "v2_chunk", "fixture": "period_episode"}}), encoding="utf-8") now = cr.git_tip()["tip"] monkeypatch.setattr(cr, "git_tip", lambda: {"tip": now, "clean": True, "dirty_files": []}) monkeypatch.setattr(cr, "recorded_tip", lambda run_id: "a" * 40) monkeypatch.setattr(cr, "fixture_dimensions", lambda f: {"d": 1}) row = {"kind": cp.EVENT_CODE_TRANSITION, "from_tip": "a" * 40, "to_tip": now, "scenario": {k: self.SC[k] for k in cr._SCENARIO_AXES}, "approved": cr.approved_for("v2_chunk"), "dimensions": {"d": 1}, "locks": dict(cr.LOCKED_CONTRACT)} first = cp.append_event(d, dict(row)) # ★approved_images 없음(옛 줄) # 옛 줄은 승인 0 으로 읽힌다 — 이 범위는 80 이 승인돼 있어 그 줄로는 못 잇는다 with pytest.raises(cr.ScopeMismatch, match="approved_images"): cr.assert_resume_transition(rid, dict(self.SC)) # 승인 없는 범위(다른 target)면 옛 줄(0)로 잇는다 — 앞 줄을 대신하는 줄 하나 other = {**self.SC, "target": "world_guide"} row2 = {**row, "scenario": {k: other[k] for k in cr._SCENARIO_AXES}, "supersedes": first["event_id"]} cp.append_event(d, row2) assert cr.assert_resume_transition(rid, dict(other))["resume"] is True