"""**기록이 바뀐 것**과 **돈이 나간 것**을 가른다 (2026-09-20 Codex (w)). ## 무엇이 틀렸나 JIT 재생성 제동은 「record 묶음이 움직였나」로 지출을 잰다. 그런데 캐시 소급 재선택은 **생성도 판정도 안 사면서** `selected` 와 산출 bytes 를 바꾼다 — 그것만으로 「재생성 방문」으로 세면 **상한이 무료 행동에 소모되고** 보고도 틀린다. ## 가르는 두 축 (영구 boolean 하나가 아니다) · 이번 방문의 **구조적 선택 행동** — `gate_select_action_this_run` (`_this_run` 이라 지출 스냅샷이 걸러 낸다). 구매 표식이 **아니다**. · **유료 구간 진입** — `shot_run_spend_attempt_count` 의 움직임. 이것도 「이미지 몇 장·VLM 몇 회」가 **아니다**. ★**입증된 무료만** 좁힌다. 아직 못 재는 경로를 0 으로 치고 제동을 넓게 풀지 않는다. `gate.outcome` 으로 무료를 판정하지도 않는다 — `reselected` 라도 C 를 **유료로 사고** 기존 B 가 이긴 경우가 있다. """ from __future__ import annotations import json from contextlib import ExitStack from pathlib import Path from typing import Any, Dict, List from unittest.mock import MagicMock, patch import pytest from app.modules.pipeline.multiroll_select import roll_labels from tests.services.test_still_cine_stage import ( EID, PID, _FLAG_PATCHES, _db_mock, _still, _write_cp, ) L2 = roll_labels(2) L3 = roll_labels(3) RR = L3[-1] TAG = "S1sh1" def _judge(violations: Dict[str, List[str]], *, ranking=None): calls: List[Any] = [] def judge_fn(tag, prompt, labeled_refs, cand_paths, labels): calls.append(list(labels)) rank = list(ranking or labels) return { "winner": rank[0], "ranking": rank, "verdicts": [{"label": l, "score": 10 - rank.index(l), "verdict_ko": "ok"} for l in labels], "readings": [{"label": l, "hard_violations": violations.get(l, [])} for l in labels], } judge_fn.owns_order = True judge_fn.emits_readings = True judge_fn.calls = calls return judge_fn def _walk(tmp_path, *, initial, rejudge, records=None, done=(), reroll=True, uid="U1", caplog=None, limit=None, cine=None): from app.services.still_recipe_service import ( run_still_recipe_generation, ) _write_cp(tmp_path, "shot_ref_classify", {"shots": {TAG: {"person_visible": True}}, "scenes": {}, "world_anchor_en": ""}) _write_cp(tmp_path, "shot_continuity", {"pose_canon": []}) _write_cp(tmp_path, "shot_conti_light", {"contis": {}}) recipe = tmp_path / "scene" / "recipe" recipe.mkdir(parents=True, exist_ok=True) if records is not None: (recipe / "records.json").write_text( json.dumps(records, ensure_ascii=False), encoding="utf-8") made: List[str] = [] def _gen_fn(tag, prompt, labeled_refs, out_path: Path): made.append(out_path.stem.rsplit("_", 1)[-1]) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_bytes(f"img:{out_path.stem}".encode()) return out_path per = MagicMock() per.safety_ladder_call_provenance.return_value = None per._resolve_generation_call_id.return_value = None per.save_single_scene_asset.side_effect = lambda *a, **k: MagicMock( id="G", is_primary=1, file_path="/x/g.png") patches = [ patch("app.core.config.settings.projects_dir", str(tmp_path)), patch("app.modules.pipeline.multiroll_gemini.make_nb2_gen_fn", return_value=_gen_fn), patch("app.modules.pipeline.multiroll_gemini.make_gemini_judge_fn", side_effect=lambda **kw: ( rejudge if kw.get("step_tag") == "still_recipe_gate_rejudge" else initial)), patch("app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn", return_value=MagicMock()), patch("app.modules.llm.opik_trace.current_shot_uid", return_value=uid), ] + [patch(f"app.core.config.settings.{n}", v, create=True) for n, v in _FLAG_PATCHES] + [ patch("app.core.config.settings.still_cine_transform_enabled", cine is not None, create=True), patch("app.modules.pipeline.cine_provider.build_cine_client", return_value=cine), patch("app.core.config.settings.still_winner_gate_enabled", True, create=True), patch("app.core.config.settings.still_gate_reroll_enabled", reroll, create=True), ] + ([patch("app.core.config.settings.still_jit_regen_limit", limit, create=True)] if limit is not None else []) with ExitStack() as st: for p in patches: st.enter_context(p) run_still_recipe_generation( db=_db_mock(), project_id=PID, episode_id=EID, stills=[_still("a", 1, 1)], stills_orm=[], entity_lookup={}, ref_image_map={}, reference_svc=MagicMock(), scene_ref_image_map={}, scene_ref_asset_id_map={}, staging_map={}, scene_cp=MagicMock(), persistence_svc=per, progress=MagicMock(), project_config=None, scene_dir=tmp_path / "scene", already_done_stills=set(done), target_scenes=None) recs = json.loads((recipe / "records.json").read_text(encoding="utf-8")) return made, recs def _free_tags(text: str) -> str: """걷기 끝 요약에서 **무료로 분류된 샷 목록** 줄만 본다.""" for line in text.splitlines(): if "무료 재선택" in line and "샷**" in line: return line return "" def _spend(rec: Dict[str, Any]) -> int: return int(rec.get("shot_run_spend_attempt_count") or 0) # ── ① 무료 재선택: 생성 0 · 판정 0 인데 선택·bytes 는 바뀐다 ──────── def test_a_free_reselect_is_not_counted_as_a_regen(tmp_path, caplog): """★★소급 재선택은 **무료**다 — 상한에 세면 안 된다. 생성도 판정도 안 사는데 `selected` 와 산출 bytes 가 바뀐다. 기록이 움직였다는 이유만으로 「재생성 방문」으로 세면 상한이 **무료 행동**에 소모된다. """ # ── 1차: 게이트를 끄고 만든다(A 가 이긴다). made1, recs1 = _walk( tmp_path, initial=_judge({"A": [], "B": []}), rejudge=_judge({}), reroll=False) assert made1 == ["a", "b"] spend1 = _spend(recs1[TAG]) # ── 2차: 판정문만 「A 실격 · B 허용」으로 바꾼 채 완료 샷으로 재방문. # 소급이 **무료로** B 를 고른다. seeded = dict(recs1[TAG]) seeded["readings"] = [{"label": "A", "hard_violations": ["x"]}, {"label": "B", "hard_violations": []}] seeded["gate"] = {"outcome": "clean", "policy": "옛-정책"} with caplog.at_level("INFO"): made2, recs2 = _walk( tmp_path, initial=_judge({}), rejudge=_judge({}), records={TAG: seeded}, done={"a"}, uid="U2") assert made2 == [], "무료 재선택인데 그림을 샀다" assert recs2[TAG]["selected"] == "B", "재선택이 안 일어났다" assert _spend(recs2[TAG]) == spend1, "무료인데 유료 진입이 올랐다" act = recs2[TAG]["gate_select_action_this_run"] assert act["action"] == "reselect_cached" assert act["from"] == "A" and act["to"] == "B" assert act["materialized"] is True and act["to_sha"] assert "무료 재선택" in caplog.text, "보고에 안 남는다" def test_the_free_action_is_not_a_spend_trace(tmp_path): """★행동 기록 자체가 **지출로 읽히면 안 된다** — `_this_run` 이라 걸린다.""" from app.services.still_recipe_service import _jit_tag_snapshot class _R: def __init__(self, d): self.data = d plain = _R({TAG: {"selected": "A"}}) acted = _R({TAG: {"selected": "A", "gate_select_action_this_run": { "action": "reselect_cached", "from": "A", "to": "B", "to_sha": "s", "materialized": True}}}) assert (_jit_tag_snapshot(plain, TAG, {TAG}) == _jit_tag_snapshot(acted, TAG, {TAG})) # ── ② 유료인데 최종 bytes 가 안 바뀐 경우 ────────────────────────── def test_a_paid_reroll_that_loses_still_counts_as_spend(tmp_path): """★★C 를 **샀는데 졌다** — 최종 교체 0 이어도 지출은 남는다. `gate.outcome` 으로 무료를 판정하면 여기서 틀린다(`reselected` 인데 유료다). """ made, recs = _walk( tmp_path, initial=_judge({"A": ["x"], "B": ["y"]}), rejudge=_judge({"A": ["x"], "B": [], RR: ["z"]}, ranking=[RR, "A", "B"])) assert RR.lower() in made, "재롤을 안 샀다" assert recs[TAG]["gate"]["outcome"] == "reselected" assert recs[TAG]["gate_reroll"]["sha256"], "산 것을 기록 안 했다" # ★★행동 기록은 「무엇을 했나」이지 **무료 주장이 아니다**. 이 방문은 # 재선택을 했고(행동 기록 있음) **동시에 유료**다 — 무료 판정은 # 행동이 아니라 **지출 계수**가 한다. assert recs[TAG]["gate_select_action_this_run"]["to"] == "B" assert _spend(recs[TAG]) >= 1, "유료 진입이 기록에 없다" def test_a_settled_failure_visit_spends_nothing_more(tmp_path): """★확정 실패 재방문 — 선택·지출·추가 영속 **모두 안 오른다**.""" made1, recs1 = _walk( tmp_path, initial=_judge({l: ["x"] for l in L2}), rejudge=_judge({l: ["x"] for l in L3})) assert RR.lower() in made1 spend1 = _spend(recs1[TAG]) sel1 = recs1[TAG]["selected"] made2, recs2 = _walk( tmp_path, initial=_judge({l: ["x"] for l in L2}), rejudge=_judge({l: [] for l in L3}), records=recs1, done={"a"}, uid="U2") assert made2 == [] assert recs2[TAG]["selected"] == sel1, "종착이 뒤집혔다" assert _spend(recs2[TAG]) == spend1, "아무것도 안 샀는데 진입이 올랐다" assert "gate_select_action_this_run" not in recs2[TAG] def test_the_paid_reselect_is_not_reported_as_free(tmp_path, caplog): """★★**종착으로 무료를 판정하지 않는다** (Codex (w) 3). `reselected` 인데 C 를 **유료로 사고** 기존 B 가 이긴 방문이다. 행동 기록은 있지만 **지출 계수가 움직였으므로** 무료가 아니다. """ with caplog.at_level("INFO"): made, recs = _walk( tmp_path, initial=_judge({"A": ["x"], "B": ["y"]}), rejudge=_judge({"A": ["x"], "B": [], RR: ["z"]}, ranking=[RR, "A", "B"])) assert RR.lower() in made assert recs[TAG]["gate"]["outcome"] == "reselected" assert "무료 재선택" not in caplog.text, ( "유료로 산 방문을 무료로 보고했다 — 종착만 보고 판정한 것이다") def test_a_materialize_only_recovery_sends_nothing(tmp_path, caplog): """★★pending 물질화만 복구 — 새 발송 0 이고 **무료다**. 재롤은 「예고(`pending_pick`) → 파일 교체 → 확정」 순서다. 파일 교체 에서 끊긴 뒤의 재개가 하는 일은 **파일을 옮기고 예고를 확정으로 바꾸는 것뿐**이다 — 이미지도 판정도 안 산다. ★종전에는 이 갈래를 **과다 계상** 쪽에 뒀다. 기록에서 드러나는 변화가 `gate_reroll.pending_pick` 이 사라지는 것뿐인데 되돌릴 손잡이가 없었기 때문이다. 이제 `_materialize_reroll_pick` 이 **무엇을 이어받았는지**(`resumed_pending_pick`)를 남기고, 비교할 때 그 한 칸을 되돌린다. `gate_reroll` 을 통째로 거르지 **않는다** — 진짜 재롤의 유료 흔적이 거기 남는다. ★같은 방문에서 사고 곧바로 확정하는 길은 이 표식을 **안 남긴다**. 그쪽은 지출 계수가 이미 올라 무료로 안 떨어진다. """ import shutil as _sh import app.modules.pipeline.multiroll_select as _ms _orig = _sh.copy n = {"i": 0} def _boom(src, dst): n["i"] += 1 if n["i"] >= 2: raise OSError("디스크 가득") return _orig(src, dst) _ms.shutil.copy = _boom try: _walk(tmp_path, initial=_judge({l: ["x"] for l in L2}), rejudge=_judge({l: (["x"] if l in L2 else []) for l in L3}, ranking=[RR, "A", "B"])) finally: _ms.shutil.copy = _orig recipe = tmp_path / "scene" / "recipe" recs1 = json.loads( (recipe / "records.json").read_text(encoding="utf-8")) assert recs1[TAG]["gate_reroll"]["pending_pick"] == RR spend1 = _spend(recs1[TAG]) from app.services.still_recipe_service import JIT_LATCH_FILENAME # ★판정 대역을 **변수로 잡는다** — 「판정 0」을 진입 계수의 **대리**로 # 재지 않고 호출 목록을 직접 본다 (Codex NON-BLOCK). init2, rej2 = _judge({l: ["x"] for l in L2}), _judge({}) with caplog.at_level("INFO"): made2, recs2 = _walk( tmp_path, initial=init2, rejudge=rej2, records=recs1, done={"a"}, uid="U2", limit=1) # ★상한 1 — 한 번이라도 세면 래치 assert made2 == [], "물질화만 하는데 그림을 샀다" assert init2.calls == [] and rej2.calls == [], ( f"판정을 다시 샀다: 선정 {init2.calls} · 재판정 {rej2.calls}") assert recs2[TAG]["gate"]["outcome"] == "resolved" assert recs2[TAG]["selected"] == RR assert _spend(recs2[TAG]) == spend1, "발송이 없는데 유료 진입이 올랐다" # ★이어받은 것을 기록에 남겨야 되돌려 견줄 수 있다 assert recs2[TAG]["gate_select_action_this_run"][ "resumed_pending_pick"] == RR # ★★**무료로 분류돼야 한다** — 상한이 무료 행동에 소모되면 안 된다 assert TAG in _free_tags(caplog.text), ( "★예고를 이어받기만 한 방문을 **유료로 셌다** — 상한이 깎인다") # ★계수는 **래치 파일**로 본다. 요약 줄을 훑으면 **줄이 아예 없을 때도 # 0 으로 읽혀서**, 요약을 지우거나 형식만 바꿔도 통과한다 # (Codex NON-BLOCK). latch = tmp_path / "scene" / "recipe" / JIT_LATCH_FILENAME assert not latch.exists(), ( "무료 재개가 상한을 소모해 **래치가 걸렸다**") def test_a_paid_visit_through_the_jit_gate_is_never_reported_free( tmp_path, caplog): """★★**완료 샷 재방문에서 유료로 샀는데** 무료로 보고하면 안 된다. 여기가 위험한 자리다 — 이 방문은 JIT 제동 판정을 **실제로 지나고**, 재선택 행동 기록도 남긴다. 행동 기록만 보고 무료로 치면 **상한이 유료 방문을 안 센다**. """ # 1차: 깨끗하게 끝낸다(게이트 대상이지만 위반 없음). made1, recs1 = _walk( tmp_path, initial=_judge({"A": [], "B": []}), rejudge=_judge({}), reroll=False) assert made1 == ["a", "b"] spend1 = _spend(recs1[TAG]) # 2차: 완료 샷으로 재방문 — 판정문이 「둘 다 실격」이라 **재롤을 산다**. # 재판정은 기존 B 를 고른다(재선택 행동 + 유료). seeded = dict(recs1[TAG]) seeded["readings"] = [{"label": l, "hard_violations": ["x"]} for l in L2] seeded["gate"] = {"outcome": "clean", "policy": "옛-정책"} with caplog.at_level("INFO"): made2, recs2 = _walk( tmp_path, initial=_judge({}), rejudge=_judge({"A": ["x"], "B": [], RR: ["z"]}, ranking=[RR, "A", "B"]), records={TAG: seeded}, done={"a"}, uid="U2") assert RR.lower() in made2, "재롤을 안 샀다(이 반례가 무의미)" assert recs2[TAG]["gate_select_action_this_run"]["to"] == "B", ( "재선택 행동 기록이 없다(이 반례가 무의미)") assert _spend(recs2[TAG]) > spend1, "유료로 샀는데 진입이 안 올랐다" assert "무료 재선택" not in caplog.text, ( "★유료로 산 방문을 **무료로 보고**했다 — 행동 기록만 보고 " "판정하면 상한이 유료 방문을 안 센다") # ── 로그가 아니라 **계수·래치**를 잰다 (Codex BLOCK 1) ───────────── def test_a_free_reselect_does_not_trip_the_latch(tmp_path): """★★**보고만 무료**이고 상한은 그대로 오르던 자리. `_records_changed=False` 만 만들면, bytes 가 달라졌을 때 「동일 산출」 갈래를 **둘 다 건너뛰고** 무조건 증가에 도착한다 — 「무료 재선택」과 「입력이 낡아 재생성」을 같은 방문에 보고하고 **래치까지** 건다. ★로그 한 문장으로 계수 확인을 대신하지 않는다 — **래치 파일**을 본다. """ from app.services.still_recipe_service import JIT_LATCH_FILENAME made1, recs1 = _walk( tmp_path, initial=_judge({"A": [], "B": []}), rejudge=_judge({}), reroll=False) assert made1 == ["a", "b"] seeded = dict(recs1[TAG]) seeded["readings"] = [{"label": "A", "hard_violations": ["x"]}, {"label": "B", "hard_violations": []}] seeded["gate"] = {"outcome": "clean", "policy": "옛-정책"} made2, recs2 = _walk( tmp_path, initial=_judge({}), rejudge=_judge({}), records={TAG: seeded}, done={"a"}, uid="U2", limit=1) # ★상한 1 — 한 번이라도 세면 래치 assert made2 == [], "무료인데 그림을 샀다" assert recs2[TAG]["selected"] == "B" latch = tmp_path / "scene" / "recipe" / JIT_LATCH_FILENAME assert not latch.exists(), ( "무료 재선택이 상한을 소모해 **래치가 걸렸다**") def test_a_third_unchanged_visit_reports_no_free_action(tmp_path, caplog): """★★`_this_run` 이 **이름뿐**이라 다음 방문이 물려받던 자리. 한 번 재선택한 샷이 **무변경 방문마다** 「이번 무료 재선택」으로 집계되면, 그 방문이 변환만 다시 시도할 때 유료 흔적까지 지운다. ★**범위**(Codex NON-BLOCK): `_walk` 의 DB·저장 대역은 **상태를 이어 가지 않는다**(고정 자산을 돌려준다). 그래서 이 시험을 「추가 자산 저장 0」의 증명으로 넓혀 쓰지 않는다 — 여기서 잠그는 것은 **행동 표식 부재와 무료 보고 0** 이다. """ made1, recs1 = _walk( tmp_path, initial=_judge({"A": [], "B": []}), rejudge=_judge({}), reroll=False) seeded = dict(recs1[TAG]) seeded["readings"] = [{"label": "A", "hard_violations": ["x"]}, {"label": "B", "hard_violations": []}] seeded["gate"] = {"outcome": "clean", "policy": "옛-정책"} _, recs2 = _walk(tmp_path, initial=_judge({}), rejudge=_judge({}), records={TAG: seeded}, done={"a"}, uid="U2") assert recs2[TAG]["gate_select_action_this_run"]["to"] == "B" # ── 세 번째: 아무것도 안 바뀐 방문이다. # ★앞 방문의 로그를 지운다 — 안 지우면 2차의 「무료 재선택」이 # 3차 것으로 읽혀 이 반례가 무의미해진다. caplog.clear() with caplog.at_level("WARNING"): made3, recs3 = _walk( tmp_path, initial=_judge({}), rejudge=_judge({}), records=recs2, done={"a"}, uid="U3") assert made3 == [] assert "gate_select_action_this_run" not in recs3[TAG], ( "옛 방문의 행동 표식을 물려받았다 — 무변경 방문이 「이번 무료 " "재선택」으로 집계된다") assert "무료 재선택" not in caplog.text # ── 변환이 유료로 돌면 무료로 덮이면 안 된다 (Codex BLOCK 2) ─────── def test_a_paid_cine_is_not_erased_by_a_free_reselect(tmp_path, caplog): """★★**변환 지출을 본체 행동 하나로 지우지 않는다**. `_jit_tag_snapshot` 은 자식·공유까지 묶는데, 본체 계수 불변만 보고 묶음 전체를 무료로 덮으면 **변환이 유료로 돌았는데도** 지워진다. ★★**이 시험이 잠그는 범위**(Codex NON-BLOCK): 변환이 **성공**으로 응답하고 둘째 입력에서 옛 `::cine` 기록을 빼 둔 판이다. 그래서 확인하는 것은 **「신규 cine 호출이 있는 방문을 무료 목록에서 제외한다」**까지다. **넓혀 쓰지 않는다** — 「기존 cine 기록을 보존한 재개에서 기각·실패 + 최종 bytes 동일이어도 래치·메타가 정확하다」는 **이 시험이 잠근 것이 아니다**. 그 갈래는 후속 범위다. """ import io from PIL import Image buf = io.BytesIO() Image.new("RGB", (8, 8), (7, 7, 7)).save(buf, format="PNG") png = buf.getvalue() calls: List[str] = [] class _Client: def set_context(self, **kw): pass def generate_image(self, prompt, labeled_references=None, **kw): calls.append(prompt) return png, 10 # ★**두 방문 다 변환을 켠다** — 한쪽만 켜면 지문이 달라져 초기 # 후보부터 다시 사고, 그러면 이 반례가 무의미해진다. made1, recs1 = _walk( tmp_path, initial=_judge({"A": [], "B": []}), rejudge=_judge({}), reroll=False, cine=_Client()) assert made1 == ["a", "b"] calls.clear() seeded = dict(recs1[TAG]) seeded["readings"] = [{"label": "A", "hard_violations": ["x"]}, {"label": "B", "hard_violations": []}] seeded["gate"] = {"outcome": "clean", "policy": "옛-정책"} with caplog.at_level("WARNING"): made2, recs2 = _walk( tmp_path, initial=_judge({}), rejudge=_judge({}), records={TAG: seeded}, done={"a"}, uid="U2", cine=_Client()) caplog_text = caplog.text assert made2 == [], "재선택은 그림을 안 사야 한다" assert calls, "변환이 안 돌았다(이 반례가 무의미)" assert recs2[TAG]["gate_select_action_this_run"]["to"] == "B", ( "재선택이 안 일어났다(이 반례가 무의미)") # ★★변환이 **유료로 돌았으므로** 이 방문은 무료가 아니다 — # `{tag}::cine` 자식 기록이 움직였으니 무료 갈래로 오면 안 된다. assert f"{TAG}::cine" in recs2, "변환 기록이 없다" assert TAG not in _free_tags(caplog_text), ( "★변환이 유료로 돌았는데 **본체 행동 하나로 무료로 덮었다**")