"""fixture 주행의 **3층 상한표**. ★아무것도 안 산다. Codex 가 층을 못박았다 (2026-08-31) — > ①step/model 별 **logical plan**: reused/cache hit/new dispatch 분리 > ②**counted transmission cap**: 예산이 실제 reserve 하는 provider 전송 수 > ③**raw HTTP upper bound**: key-slot·router/fallback·SDK 내부 retry 까지 > 검색 query/다운로드/이미지 검색/이미지 생성도 **각각 별도 열**입니다. > reserve 가 없는 유료 경로는 **상한을 주장하지 말고** 먼저 문을 달거나 > **「미측정」으로 막으십시오.** ## ★글 호출에는 세는 문이 없다 — 지어내지 않는다 `grounding_chunk_step.physical_per_logical` 이 이미 적어 두었다 — > 글 호출 경로(`_completion`)에는 이미지 쪽 `reserve_current_call` 같은 > **세는 자리가 없다.** 그래서 물리 수를 「막을」 수는 없고, 대신 **구조적으로 > 상한을 정한다**. 그래서 이 표는 층②에 글 스텝을 **`미측정`** 으로 적는다. 그 대신 **스텝 경계**에 문을 단다 — 한 스텝이 제 논리 상한을 넘으면 **다음 스텝 앞에서** 선다. 한 스텝 **안에서** 폭주하는 것은 이 문으로 못 막는다는 것도 같이 적는다. """ from __future__ import annotations from pathlib import Path from typing import Any, Dict, List, Optional #: ★★★`cwd` 가 아니라 **모듈 자리**에서 뽑는다 (실측 2026-09-01). #: 앞에는 `Path.cwd()` 기준 상대 경로를 냈다. 저장소 뿌리에서 부르면 그 길이 #: 안 열려 `boundary_of` 가 **문을 못 찾고**, 유료 스텝 30여 개가 조용히 #: **「무료」로 분류됐다**. 재는 도구가 어디서 부르냐에 따라 다른 답을 내면 #: 그것으로 잰 수는 아무 뜻이 없다. BACKEND = Path(__file__).resolve().parents[2] class WrongWorkingDirectory(RuntimeError): """`backend/` 밖에서 불렸다. ★아무것도 안 하고 선다.""" def settings_came_from_backend_env() -> bool: """`settings` 가 **`backend/.env`** 를 읽었나. ★★★pydantic-settings 는 `.env` 를 **작업 디렉토리 기준**으로 찾는다. 저장소 뿌리에서 부르면 그 파일을 못 찾아 **전부 기본값**이 된다 — 실측 2026-09-01: `background_mode` 가 True 대신 False 로 읽혀 유료 스텝 여섯이 통째로 「건너뜀」이 됐다. 재는 도구가 어디서 부르냐에 따라 다른 답을 내면 그 수는 아무 뜻이 없다. """ return (Path.cwd().resolve() == BACKEND and (BACKEND / ".env").is_file()) def assert_backend_cwd() -> None: """★유료 입구에서 **먼저** 부른다.""" if not settings_came_from_backend_env(): raise WrongWorkingDirectory( f"`backend/` 에서 불러야 한다 (지금 {Path.cwd()}). " "여기서는 `.env` 를 못 읽어 settings 가 전부 기본값이 되고, " "유료 스텝이 조용히 「무료·건너뜀」으로 분류된다") #: 유료 갈래 — ★각각 **별도 열**이다 (Codex). KIND_TEXT = "text_llm" KIND_SEARCH = "search_query" KIND_DOWNLOAD = "download" KIND_IMAGE_SEARCH = "image_search" KIND_IMAGE_GEN = "image_generation" #: ★★글 호출의 **중앙 문 둘**. 앞 판은 「문이 없다」고 적었는데 반대였다 #: (Codex BLOCK 1 2026-08-31). 물리 전송 자리에 이미 걸려 있고, 팔을 안 #: 들었을 뿐이다. CENTRAL_DOORS = ("app/modules/llm/llm_client.py:438 " "(router.completion 직전)", "app/core/openai_keys.py:394 (OpenAI SDK 호출 직전)") #: 세는 문이 **있는** 갈래. ★없는 것은 상한을 주장하지 않는다. COUNTED = { KIND_TEXT: CENTRAL_DOORS, KIND_IMAGE_GEN: "app.core.image_call_budget.reserve_current_call", KIND_SEARCH: "app.core.research_call_budget.reserve_current_research_call", } UNMEASURED = "미측정 — 세는 문이 없다" class CapExceeded(RuntimeError): """센 것이 승인된 상한을 넘었다. ★다음 스텝 앞에서 선다.""" def slots_for(alias: str, *, provider: Optional[str] = None ) -> Dict[str, Any]: """그 모델이 도는 **키 슬롯 수**. ★못 세면 「모른다」로 적는다. ★★★alias 앞글자로 가르면 틀린다 (실측 2026-09-02): `gpt-mini` 는 이름은 gpt 인데 **물리 모델이 gemini** 라 gemini 키 풀로 돈다. 앞 판은 그것을 openai 슬롯으로 셌다 — 이번엔 둘 다 2 라 수가 같았을 뿐이다. manifest 의 `provider` 를 우선 보고, 없으면 **물리 모델**로 가른다. """ try: prov = str(provider or "").strip().lower() if not prov: prov = "gemini" if str(_physical_of(alias)).startswith( "gemini") else "openai" if prov == "gemini": from app.modules.llm.gemini_key_pool import key_count return {"slots": int(key_count()), "how": "gemini_key_pool"} from app.core.openai_keys import slot_count return {"slots": int(slot_count()), "how": "openai_keys"} except Exception as exc: # noqa: BLE001 return {"slots": None, "how": f"못 셌다: {type(exc).__name__}"} def steps_to(target: str = "scene_detail") -> List[str]: """`target` 까지 **의존 그래프에서 뽑은** 유료 스텝. ★손으로 안 적는다.""" from app.core.step_manifest import STEP_MANIFEST as M seen: set = set() def walk(sid: str) -> None: if sid in seen or sid not in M: return seen.add(sid) for d in (M[sid].get("depends_on") or ()): walk(d) walk(target) # ★`provider='-'` 는 **모델을 안 부르는 코드 스텝**이다 (`scene_save`). # 그것까지 유료로 세면 상한이 실제보다 커져 승인이 헐거워진다. paid = [s for s in seen if str(M[s].get("provider") or "-").strip() not in ("", "-")] return sorted(paid, key=lambda s: M[s].get("order") or 0) def physical_upper(*, logical: int, contract: Dict[str, Any], slots: Optional[int]) -> Dict[str, Any]: """층②③. ★**공용 계약**을 부른다 — 식을 여기 다시 안 적는다. ★앞 판은 `physical_per_logical` 만 불러 **counted 를 raw 라고 적었다**. litellm 이 제 client 에 주는 `max_retries` 가 `reserve` **아래**라 raw 는 counted 의 최대 3배다 (Codex BLOCK 2 · `ref_canary` 가 이미 기록). """ from tools.grounding_audit.call_bound_contract import bounds return bounds(logical=logical, contract=contract, slots=slots) def plan_rows(*, logical_caps: Dict[str, int], contract: Dict[str, Any], target: str = "scene_detail") -> List[Dict[str, Any]]: """스텝마다 세 층을 **나란히** 적는다. Args: logical_caps: 스텝 → **손으로 승인한** 논리 상한. ★층② `counted_door` 는 **경로마다 다르다** — 중앙 둘에는 문이 있고 직접 경로에는 없다. 「다 없다」로 적었던 것을 바로잡았다 (Codex BLOCK 1). """ from app.core.step_manifest import STEP_MANIFEST as M rows = [] for s in steps_to(target): alias = str(M[s].get("default_model") or "") sl = slots_for(alias, provider=M[s].get("provider")) cap = int(logical_caps.get(s, 0)) b = physical_upper(logical=cap, contract=contract, slots=sl["slots"]) rows.append({ "step": s, "order": M[s].get("order"), "kind": KIND_TEXT, "model_alias": alias, "model_physical": _physical_of(alias), "fan_out": bool(M[s].get("fan_out")), # ① "logical_cap": cap, "logical_new": cap, "logical_reused": 0, "logical_cache_hit": 0, # ② ★문은 **있다** — canary 범위에서 팔을 들면 세어진다 "counted_door": CENTRAL_DOORS, "counted_cap": b["counted"], # ③ ★막을 수 **없는** 상한 "key_slots": sl["slots"], "slots_how": sl["how"], "contract": dict(contract), "raw_upper": b["raw_http"], "layers": b["layers"], }) return rows def _physical_of(alias: str) -> Optional[str]: """alias → 실제 모델 이름. ★production 해석기를 쓴다.""" try: # ★production 해석기 한 벌 (`era_research` 가 쓰는 것과 같은 것) from app.modules.pipeline.era_research import resolve_model_physical return str(resolve_model_physical(alias)) except Exception as exc: # noqa: BLE001 return f"못 풀었다: {type(exc).__name__}" def totals(rows: List[Dict[str, Any]]) -> Dict[str, Any]: raw = [r.get("raw_upper") for r in rows] cnt = [r.get("counted_cap") for r in rows] return { "steps": len(rows), "logical_cap_total": sum(r["logical_cap"] for r in rows), "counted_cap_total": (None if any(x is None for x in cnt) else sum(cnt)), "raw_upper_total": (None if any(x is None for x in raw) else sum(raw)), "unmeasured_steps": [r["step"] for r in rows if r.get("counted_door") == UNMEASURED], "★note": ("`counted` 는 **막을 수 있는** 수다 — canary 범위에서 " "`canary_text_scope` 가 그 자리에서 세고 넘으면 세운다. " "`raw_http` 는 그 **아래**에서 litellm 이 더 도는 몫까지 " "친 것이라 **막을 수 없는** 상한이다."), } def assert_within(*, observed: Dict[str, int], approved: Dict[str, int]) -> None: """★**다음 스텝 앞에서** 선다. 넘은 것을 조용히 줄이지 않는다.""" over = {s: (n, approved.get(s)) for s, n in observed.items() if s not in approved or n > int(approved[s])} if over: raise CapExceeded( "논리 상한을 넘었다 — 다음 스텝을 시작하지 않는다: " + ", ".join(f"{s}: {n} > 승인 {a}" for s, (n, a) in sorted(over.items()))) # ───────────────────────────────────────────────────────────────────── # ★★★「42개」는 **실행계획이 아니다** (Codex BLOCK 3, 2026-08-31) # # > `steps_to` 는 dependency closure 에서 manifest `provider != '-'` 만 봅니다. # > runtime applicability/mode/cache/reuse 와 실제 provider boundary 는 안 # > 봅니다. 42가 보수적 emergency 목록일 수는 있어도 **「실제 유료 목록」이나 # > acceptance 분모로 쓰면 안 됩니다.** # ───────────────────────────────────────────────────────────────────── #: 글이 나가는 **경로 셋**. ★앞의 둘에만 문이 있다. BOUNDARY_ROUTER = "llm_client._completion → router.completion (문 있다)" BOUNDARY_SDK = "openai_keys._invoke → OpenAI SDK (문 있다)" BOUNDARY_DIRECT = "openai_keys.llm_completion → litellm.completion (★문 없다)" #: ★모델을 **아예 안 부르는** 스텝. manifest 에 모델이 적혀 있어도 그렇다 #: (실측: `shot_dependency`·`floor_plan_overlay_payload`· #: `episode_reference_policy` 는 결정적이다). NO_LLM = "no_llm" #: 문 없는 경로를 쓰는 provider 모듈. ★여기 닿는 스텝은 **미측정**이다. _DOORLESS_HINT = "llm_completion as _llm_completion" def step_module(step: str) -> Optional[str]: """`STEP_CLASSES` 로 **실제 모듈**을 푼다. ★이름 맞추기로 안 한다.""" import inspect import app.core.steps as S reg = next((v for n in dir(S) if isinstance(v := getattr(S, n), dict) and "background_prompt" in v), None) cls = (reg or {}).get(step) if cls is None: return None f = inspect.getsourcefile(cls) if not f: return None try: return str(Path(f).resolve().relative_to(BACKEND)) except ValueError: return f def boundary_of(step: str, *, depth: int = 4) -> Dict[str, Any]: """그 스텝이 **어느 문**을 지나나. ★★★`counted` 는 넷이다 — `True` 중앙 문을 지나는 이름을 **봤다** `False` 문 없는 직접 경로에 닿는다 → **미측정** `NO_LLM` 모델을 부르는 이름이 **하나도 없다** → 유료가 아니다 `None` 어느 쪽도 못 봤다 → **모른다**. 통과 근거로 안 쓴다 ★`NO_LLM` 은 「못 찾았다」와 다르다. 모델 호출 **부류 전체**를 훑어 하나도 없을 때만 준다 — `shot_dependency` 처럼 manifest 에는 모델이 적혀 있어도 실제로는 결정적인 스텝이 있다 (실측 2026-08-31). ★그래도 **주행이 확인한다** — 그 스텝에서 counted 가 하나라도 나오면 이 판정이 틀린 것이므로 드러나야 한다. ★**AST 로** 본다. 글자로 `이름(` 만 찾으면 `fn=call_structured` 처럼 **넘겨주는** 자리를 놓친다. """ import ast from pathlib import Path mod = step_module(step) if mod is None: return {"boundary": None, "module": None, "counted": None, "why": "STEP_CLASSES 에서 클래스를 못 찾았다 — 모른다"} #: 중앙 문을 지나는 이름. ★각각 문에 닿는 것을 확인하고 넣었다. CENTRAL = {"call_structured", "router_completion", "_completion", "openai_client", "call_structured_with_retry", "call_text"} #: 문 없는 직접 경로의 이름. DOORLESS = {"llm_completion"} #: **모델을 부르는 부류 전체**. 하나도 없으면 유료가 아니다. ANY_LLM = CENTRAL | DOORLESS | { "litellm", "completion", "acompletion", "OpenAI", "AsyncOpenAI", "generate_image", "GeminiImageClient", "ReveImageClient", "GrokImageClient", "call_llm", "call_vlm", "dual_vlm", "generate_structured", "invoke_model"} seen: set = set() def scan(path: str, d: int) -> Dict[str, bool]: got = {"doorless": False, "central": False, "any_llm": False} if path in seen or d < 0: return got seen.add(path) p = Path(path) if not p.is_absolute(): p = BACKEND / p # ★어디서 불러도 같은 파일을 연다 if not p.is_file(): return got try: tree = ast.parse(p.read_text(encoding="utf-8", errors="replace")) except SyntaxError: return got names: set = set() follow: set = set() for n in ast.walk(tree): if isinstance(n, ast.Name): names.add(n.id) elif isinstance(n, ast.Attribute): names.add(n.attr) elif isinstance(n, ast.ImportFrom): for a in n.names: names.add(a.name) mod_name = str(n.module or "") if mod_name.startswith(("app.modules.pipeline.", "app.modules.llm.")): follow.add(mod_name) elif isinstance(n, ast.Import): for a in n.names: names.add(a.name.split(".")[-1]) got["doorless"] = bool(names & DOORLESS) got["central"] = bool(names & CENTRAL) got["any_llm"] = bool(names & ANY_LLM) for m in sorted(follow): sub = scan(m.replace(".", "/") + ".py", d - 1) for k in got: got[k] = got[k] or sub[k] return got r = scan(mod, depth) if r["doorless"]: return {"boundary": BOUNDARY_DIRECT, "module": mod, "counted": False, "why": "문 없는 직접 경로에 닿는다 — 이 스텝은 **미측정**이다"} if r["central"]: return {"boundary": BOUNDARY_ROUTER, "module": mod, "counted": True, "why": "중앙 문을 지나는 이름을 **실제로 봤다**(쓰거나 넘기거나)"} if not r["any_llm"]: return {"boundary": None, "module": mod, "counted": NO_LLM, "why": f"깊이 {depth} 까지 **모델을 부르는 이름이 하나도 없다** " "— 결정적 스텝이다. ★주행에서 counted 가 나오면 이 " "판정이 틀린 것이므로 드러난다"} return {"boundary": None, "module": mod, "counted": None, "why": f"모델 호출은 있는데 깊이 {depth} 까지 **어느 문인지 못 봤다** " "— 「못 찾은 것」은 「없다」가 아니다. 사람이 정해야 한다"} class ApplicabilityContradiction(RuntimeError): """선언한 설정과 **production 술어**의 답이 다르다. ★아무것도 안 하고 선다.""" def _needs_a_runner(fn: Any) -> bool: """이 술어가 **runner 를 보는가**. ★AST 로 본다 — 글자로 안 찾는다. ★★★왜 예외로는 못 가리나 (실측 2026-09-01): `_if_has_outlooks` 와 `_if_planning_doc` 은 **속으로 `except` 를 잡고 `False` 를 돌려준다**. 그러면 「모른다」가 **「아니다」로 둔갑**해서 조용히 계획이 틀어진다 — 그 스텝이 「건너뜀·0원」이 되고, 문에는 `cap=0` 이 걸린다. """ import ast import inspect import textwrap try: tree = ast.parse(textwrap.dedent(inspect.getsource(fn))) except (OSError, TypeError, SyntaxError): return True # ★못 읽으면 **모른다** 쪽으로 접는다 return any(isinstance(n, ast.Name) and n.id == "runner" for n in ast.walk(tree)) def _production_says(rule: str) -> Optional[bool]: """production 술어에게 **직접 묻는다**. runner 가 필요하면 `None`. ★runner 를 보는 술어에는 **아예 안 묻는다** — 물으면 속으로 예외를 삼키고 `False` 를 주는 것이 있어서, 「모른다」가 「아니다」가 된다. """ from app.core.applicability import APPLICABILITY_VALIDATORS fn = APPLICABILITY_VALIDATORS.get(rule) if fn is None or _needs_a_runner(fn): return None try: return bool(fn(None)) # type: ignore[arg-type] except Exception: # noqa: BLE001 return None def execution_plan(*, config: Dict[str, Any], target: str = "scene_detail", resolve: Optional[Any] = None) -> Dict[str, Any]: """**돌 스텝**을 낸다. ★`applied` 는 **실행 목록**이지 유료 목록이 아니다. ★★★실행은 **전체 closure** 에 applicability 만 건다. 비용을 세려고 거른 목록을 실행에 쓰면 `scene_save` 같은 **코드 스텝이 빠져** 뒤 스텝이 제 체크포인트 없이 돈다 (Codex 실측 2026-08-31: closure 43 · 거른 것 42). Args: config: fixture 의 **선언된** 설정 — `{"grounding_v2": False, ...}`. resolve: `(step, manifest_row, config) -> bool|None`. 안 주면 선언 기준으로 본다. ★`None` 은 「모른다」다 — 「적용된다」로 접지 않는다. 문 판정도 마찬가지다. """ from app.core.step_manifest import STEP_MANIFEST as M def _declared(step, row, cfg): """★★★**production 술어를 직접 부른다** — 내가 지은 dict 로 안 푼다. 실측 2026-09-01 — 앞 판은 `cfg` 에 적은 값을 그대로 믿었다. 그런데 `if_visual_continuity_anchor_enabled` 는 **`settings`** 를 보고, `.env` 에 `VISUAL_CONTINUITY_ANCHOR_ENABLED=true` 가 있었다. 즉 계획표는 「적용 안 됨·0원」이라 적고 pipeline 은 **실제로 사러 갔다**. 게이트가 잡아서 판이 섰지, 어긋난 것은 pipeline 이 아니라 **재는 쪽** 이었다. always → True disabled → False on_demand → True (수동 호출 전제) if_* → `APPLICABILITY_VALIDATORS` 의 **그 함수**를 부른다. runner 가 있어야 답하는 것은 **모른다**로 둔다. ★`cfg` 는 이제 **대조용**이다 — 선언과 실제가 다르면 **선다**. 조용히 한쪽을 고르면 「내가 원하는 답을 산출에 심는 것」이 된다. """ rule = str(row.get("applicability") or "always") if rule in ("always", "on_demand"): return True if rule == "disabled": return False if not rule.startswith("if_"): return None # ★모르는 규칙을 참으로 안 접는다 key = rule[3:] live = _production_says(rule) if live is None: return bool(cfg[key]) if key in cfg else None if key in cfg and bool(cfg[key]) != live: raise ApplicabilityContradiction( f"{step}: fixture 는 {key}={bool(cfg[key])} 라고 적었는데 " f"production 술어({rule})는 {live} 라고 답한다 — 계획표가 " "거짓말을 하고 있다. 사람이 어느 쪽인지 정해야 한다") return live fn = resolve or _declared applied, skipped, unknown, blocked = [], [], [], [] free, metered, paid_unknown = [], [], [] for s in execution_steps_to(target): row = M[s] got = fn(s, row, config) code_only = is_code_step(s) b = ({"boundary": None, "module": step_module(s), "counted": NO_LLM, "why": "manifest 의 `provider` 가 '-' 다 — 코드 스텝이다"} if code_only else boundary_of(s)) rec = {"step": s, "order": row.get("order"), "applicability": row.get("applicability"), "model_alias": row.get("default_model"), "code_step": code_only, "boundary": b["boundary"], "module": b["module"], "counted": b.get("counted"), "boundary_why": b["why"]} if got is None: unknown.append(rec) elif not got: skipped.append(rec) else: applied.append(rec) c = b.get("counted") if c == NO_LLM: free.append(rec) # ★모델을 아예 안 부른다 elif c is True: metered.append(rec) # ★문을 지나므로 세어진다 else: paid_unknown.append(rec) blocked.append(rec) # ★모르는 것을 통과시키지 않는다 return { "target": target, "config": dict(config), "closure_total": len(execution_steps_to(target)), "conservative_total": len(steps_to(target)), "applied": applied, "skipped": skipped, "unknown": unknown, "blocked_no_or_unknown_door": blocked, # ★★`applied` 는 **실행 목록**이다. 유료는 `applied_metered` 뿐이다. "applied_free": [r["step"] for r in free], "applied_metered": [r["step"] for r in metered], "applied_unknown": [r["step"] for r in paid_unknown], "★means": ("`applied` 는 **이번 판에 돌 스텝 전부**다 — 코드 스텝도 " "들어간다. 유료는 `applied_metered` 뿐이고 " "`conservative_total` 은 **비용 emergency 목록**이지 " "실행 목록도 분모도 아니다. `unknown` 이 남아 있으면 " "주행 전에 정해야 한다. `blocked_no_or_unknown_door` 가 " "비어 있지 않으면 그 스텝은 **돌리지 않는다**."), "★free_note": ("`applied_free` 는 모델을 부르는 이름이 **하나도 없는** " "스텝이다(코드 스텝 포함). manifest 에 모델이 적혀 " "있어도 그렇다 — 빼지 않고 **그대로 돌린다**. 예산은 " "주행 내내 팔을 든 채로 두어, 뜻밖의 전송이 생기면 그 " "스텝의 delta 에 **즉시 잡힌다**."), "reused": 0, "cache_hit": 0, "new": len(metered), "★reuse_note": ("첫 주행이라 되쓰기·캐시가 0 이라고 **선언**한 것이다. " "주행 뒤 체크포인트로 **실측**해 다시 적는다"), } # ───────────────────────────────────────────────────────────────────── # ★★★**스텝 목록 밖에도 유료가 있다** (2026-08-31 실측) # # Codex: 「provider manifest 표시는 비용의 증거가 아닙니다」. 반대도 참이다 — # **manifest 에 없는 유료**가 있다. fixture 를 세우는 길을 따라가 보니 — # # ProjectService.create_project → generate_english_name → `call_text` # (`project_summary` · gpt) **유료 1회** # EpisodeService.create_episode → PyMuPDF 로 글만 뽑는다 **무료** # # 이 한 번을 상한에 안 넣으면 승인 밖 호출이 조용히 나간다. # ───────────────────────────────────────────────────────────────────── #: fixture 를 세울 때 **스텝 밖에서** 나가는 유료 호출. BOOTSTRAP_PAID = ( {"where": "ProjectService.create_project → generate_english_name", "file": "app/services/project_service.py:31", "step": "project_summary", "model_alias": "gpt", "logical": 1, "why": "한국어 프로젝트명을 영어로 옮긴다. 실패하면 romanization 으로 " "떨어지지만 **먼저 부른다**"}, ) def bootstrap_rows(*, contract: Dict[str, Any]) -> List[Dict[str, Any]]: """부트스트랩의 유료 행. ★스텝 표와 **같은 층 계약**으로 센다.""" out = [] for b in BOOTSTRAP_PAID: sl = slots_for(b["model_alias"]) got = physical_upper(logical=b["logical"], contract=contract, slots=sl["slots"]) out.append({**b, "kind": KIND_TEXT, "key_slots": sl["slots"], "counted_door": CENTRAL_DOORS, "counted_cap": got["counted"], "raw_upper": got["raw_http"]}) return out def per_step_delta(before: Dict[str, int], after: Dict[str, int] ) -> Dict[str, int]: """스텝 하나가 **실제로 몇 번** 보냈나. ★`free` 판정을 여기서 잠근다.""" return {"counted": int(after.get("used", 0)) - int(before.get("used", 0)), "denied": int(after.get("denied", 0)) - int(before.get("denied", 0))} def assert_free_step_sent_nothing(step: str, delta: Dict[str, int]) -> None: """★`applied_free` 로 적은 스텝이 **보냈으면** 판정이 틀린 것이다.""" if int(delta.get("counted", 0)) != 0: raise CapExceeded( f"{step} 은 무료로 분류했는데 **{delta['counted']}번 보냈다** — " "분류가 틀렸다. 여기서 선다") # ───────────────────────────────────────────────────────────────────── # ★★★**실행 목록**과 **비용 목록**은 다르다 (Codex BLOCK 2026-08-31) # # > `steps_to()` 가 dependency closure 를 만든 뒤 `provider == "-"` 를 # > 제거합니다. 실측: full closure 43, filtered 42. 빠진 것은 # > **`scene_save`(order 6, always, provider '-')** 1개입니다. # > fresh fixture 에서 `scene_save` CP 없이 뒤 의존 스텝을 시작합니다. # # 즉 **비용을 세려고 거른 목록을 실행에 쓰면 production 이 깨진다.** # 실행은 **전체 closure** 에 applicability 만 걸고, 비용은 그 **위에서** 가른다. # ───────────────────────────────────────────────────────────────────── def execution_steps_to(target: str = "scene_detail") -> List[str]: """**전체** 의존 닫힘. ★`provider` 로 거르지 않는다 — 실행 목록이다.""" from app.core.step_manifest import STEP_MANIFEST as M seen: set = set() def walk(sid: str) -> None: if sid in seen or sid not in M: return seen.add(sid) for d in (M[sid].get("depends_on") or ()): walk(d) walk(target) return sorted(seen, key=lambda s: M[s].get("order") or 0) def is_code_step(step: str) -> bool: """모델을 안 부르는 **코드 스텝**인가 (`provider == '-'`).""" from app.core.step_manifest import STEP_MANIFEST as M return str((M.get(step) or {}).get("provider") or "-").strip() in ("", "-")