"""격리 fixture canary 의 **하나뿐인 입구**. ★순서가 계약이다. ①격리 DB·디렉토리가 원본이 아니고 run_id 를 갖는다 ②DB `create_database` — 관리자 연결은 maintenance DB 여야 한다 ③schema `canary_alembic` **별도 프로세스** — 연결 안에서 DB 확인 · head ④부트스트랩 프로젝트·에피소드. ★**제 상한·제 장부** (pipeline 과 안 섞는다) ⑤pipeline 38 스텝을 **한 스텝씩** — 새 예산 · per-step delta ⑥이미지 참조 묶음 1장 (★별도 승인. 이 입구는 거기까지 안 간다) ★단계마다 **파일로 내려쓴다** — 죽어도 어디까지 갔는지 남는다. ★어느 문에서든 서면 **자동으로 늘리거나 다시 사지 않는다**. python tools/grounding_audit/canary_run.py --dry python tools/grounding_audit/canary_run.py --live # ★유료 운영자가 줄 것 (코드에 안 박는다) — THEROAD_CANARY_ADMIN_DSN maintenance DB(postgres) 관리자 연결 THEROAD_CANARY_TEMPLATE_URL 접속 정보의 본 """ from __future__ import annotations import json import os import re import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Sequence BACKEND = Path(__file__).resolve().parents[2] sys.path.insert(0, str(BACKEND)) from tools.grounding_audit import canary_bootstrap as cbs # noqa: E402 from tools.grounding_audit import canary_cost_table as ct # noqa: E402 from tools.grounding_audit import canary_isolation as ci # noqa: E402 from tools.grounding_audit import canary_pipeline as cp # noqa: E402 KST = timezone(timedelta(hours=9)) #: fixture 의 **선언된** 설정. ★D 는 아직 inert 다. #: ★★★내가 **못 정하는 값은 안 적는다** (실측 2026-09-01). #: 앞 판은 `visual_continuity_anchor_enabled: False` 를 적어 두고 그 스텝을 #: 「적용 안 됨·0원」으로 계획했다. 그런데 그 규칙은 **`settings`** 를 보고 #: `.env` 에 `VISUAL_CONTINUITY_ANCHOR_ENABLED=true` 가 있었다 — pipeline 은 #: 실제로 사러 갔고 게이트가 잡아 판이 섰다. 어긋난 것은 pipeline 이 아니라 #: **재는 쪽**이었다. 이제 `if_*` 는 **production 술어에게 직접 묻고**, #: 여기 적은 값과 다르면 `ApplicabilityContradiction` 으로 선다. #: ★★이 canary 는 **legacy 판**이다 — 고증 갈래를 하나도 안 켠다. #: ★2026-09-01 D 활성화로 술어가 둘 늘었다. **명시로 적는다** — 안 적으면 #: 비용표가 그 스텝을 `unknown` 으로 두고 「돌지 안 돌지 모르는 것」이 #: 되어 상한 계산이 흔들린다. #: 고증 갈래를 정하는 **네 술어**. ★이름을 여기 한 번만 적는다 — 값은 #: production 술어에게 물어서 채운다(손으로 안 적는다). _MODE_RULES = ("if_grounding_v2", "if_chunk_producer", "if_not_chunk_producer", "if_grounding_reference") #: 시나리오의 축. ★한 곳에 적는다 — 전이·scope·산출이 같은 목록을 본다. #: `background` 는 **process 의 실제 설정**(`settings.background_mode`)에서 #: 읽고, CLI 선언은 그것과 **대조만** 한다(2026-09-02). 설정을 뒤집으면 #: 시나리오가 바뀐 것이라 전이 기록과 되쓰기 문이 그것을 본다. _SCENARIO_AXES: tuple = ("mode", "fixture", "target", "background", "still_recipe", "outdoor") #: 설정에서 오는 축 셋 — 축 이름 → 그 축을 켜는 production 술어들 · 그 술어들이 #: 읽는 settings 이름들(뒤집힘 검사가 `_config_hash` 에서 찾는 글자) · 끄는 env. #: ★★2단계(이미지) closure 에 outdoor 사슬(`if_outdoor_lane_pipe`)과 still_recipe #: 사슬(`if_still_recipe`)이 들어온다 — 전부 unverified 단위라 live 가 선다. #: 배경처럼 **process 의 실제 설정**으로 읽고 CLI 선언은 대조만 한다 #: (Codex ① 찬성 2026-09-02: 고증 canary 두 arm 의 공통 설정으로만 기록하고 #: production 기본값 변경으로 읽지 않는다). _SETTING_AXES: Dict[str, Dict[str, Any]] = { "background": { "predicates": ("if_background_mode",), "settings": ("background_mode",), "env": "BACKGROUND_MODE=off", }, "still_recipe": { "predicates": ("if_still_recipe",), "settings": ("still_recipe_mode",), "env": "STILL_RECIPE_MODE=off", }, "outdoor": { # ★셋 중 하나라도 켜져 있으면 outdoor 사슬의 어느 스텝이 돈다 — "on" "predicates": ("if_outdoor_lane_plan", "if_outdoor_lane_pipe", "if_outdoor_direct_or_map_or_lane"), "settings": ("outdoor_lane_plan_enabled", "outdoor_lane_pipe_enabled", "outdoor_direct_compose_enabled", "outdoor_map_conti_enabled"), "env": ("OUTDOOR_LANE_PLAN_ENABLED=false OUTDOOR_LANE_PIPE_ENABLED=false " "OUTDOOR_DIRECT_COMPOSE_ENABLED=false OUTDOOR_MAP_CONTI_ENABLED=false"), }, } class _AsIfNoRunner: """설정만 보는 술어에 주는 대역 — `project_config` 빈 dict, runner 아님.""" project_config: Dict[str, Any] = {} project_id = None def axis_actual(axis: str) -> str: """이 process 가 실제로 도는 그 축의 상태 — `"on"`/`"off"`. ★술어 한 곳.""" from app.core.applicability import APPLICABILITY_VALIDATORS as V spec = _SETTING_AXES[axis] return "on" if any(bool(V[p](_AsIfNoRunner())) for p in spec["predicates"]) else "off" def background_actual() -> str: """이 process 가 실제로 도는 배경 모드 — `"on"`/`"off"`. ★`axis_actual` 의 별칭.""" return axis_actual("background") def fixture_config(grounding_mode: str, fixture: Optional[str] = None) -> Dict[str, Any]: """이 판의 **선언**. ★네 값을 손으로 안 적는다 — 술어에게 묻는다. 앞 판은 `{"chunk_producer": False, ...}` 를 손으로 적었다. 그 넷은 `project_config` 만 보는 술어라 **물어볼 수 있는데도** 적어 뒀고, 모드를 바꾸면 네 줄을 같이 안 고치는 순간 계획표가 거짓말을 한다. ★`background_mode` 는 fixture 자신의 성질이라 그대로 선언한다. """ from app.core.applicability import APPLICABILITY_VALIDATORS as V class _AsIf: """★`project_config` 만 든 대역 — 이 넷은 그것만 본다.""" project_config = {"grounding_mode": grounding_mode} project_id = None # ★★배경 모드도 **술어에 묻는다** (2026-09-02). 앞 판은 `True` 를 손으로 # 적었다 — 그러면 process 가 `BACKGROUND_MODE=off` 로 떠도 계획표는 # 배경 사슬을 세고, 켜져 있으면 이미지 문(floor_plan_render)이 계획에 # 없는 채로 나간다. 술어는 `settings.background_mode` 를 읽는다. got = {"background_mode": bool(V["if_background_mode"](_AsIf()))} for rule in _MODE_RULES: fn = V.get(rule) if fn is None: raise KeyError(f"술어 {rule!r} 가 없다 — 이름이 바뀌었다") got[rule[3:]] = bool(fn(_AsIf())) # ★★runner 가 있어야 답하는 술어 셋 — 계획표는 production 에 못 묻고 # (`_production_says` → None) fixture 의 **선언**을 쓴다. 안 적으면 그 스텝은 # `unknown` 이라 live 가 선다(실측 2026-09-02: composite_image_gen · # outdoor_place_spec · outdoor_place_canon · background_share_plan 넷). # - has_outlooks: fixture 모듈의 `HAS_OUTLOOKS`(없으면 선언 안 함 → unknown) # - outdoor_direct_or_map_or_lane: outdoor 축의 실제 설정 # - background_share_plan: still_recipe ∧ settings 플래그 — runner 를 안 본다 got["outdoor_direct_or_map_or_lane"] = axis_actual("outdoor") == "on" got["background_share_plan"] = bool(V["if_background_share_plan"](_AsIf())) if fixture is not None: fx = cbs.load_fixture(fixture) if hasattr(fx, "HAS_OUTLOOKS"): got["has_outlooks"] = bool(getattr(fx, "HAS_OUTLOOKS")) return got #: 이 판이 도는 모드. ★`--mode` 로 바꾼다. DEFAULT_CANARY_MODE = "legacy" #: 중앙 참조 조사 스텝. ★이름을 계약에서 가져온다. CENTRAL_STEP = "reference_acquisition" #: 기본 모드의 **선언**. ★손으로 안 적는다 — 위 함수가 술어에게 물어서 낸다. #: ★모듈 값으로 남겨 둔다: 부르는 쪽·시험이 이 이름을 이미 쓴다. # ★★★모듈 import 때 `fixture_config()` 를 **부르지 않는다** (실측 2026-09-02 # 격리 near-miss): 배경 술어가 `settings` 를 읽으므로 import 때 부르면 # `app.core.config` 가 env 갱신 **전에** 올라와 원본 DB URL 로 굳는다. # 기본 모드의 선언은 `default_fixture_config()` 로 **그때** 묻는다. def default_fixture_config() -> Dict[str, Any]: return fixture_config(DEFAULT_CANARY_MODE) #: ★★★2026-09-02 (Codex) — **세 겹을 다 잠근다**. 앞 판은 `num_retries` #: 하나만 잠그고 tier·SDK 재시도는 열어 뒀다. #: #: num_retries Router — `settings.llm_max_retries` 를 0 으로 #: enable_fallback Tier 2/3 를 닫는다 → tier 1 #: sdk_max_retries litellm 이 제 client 에 주는 값 → 0 #: #: ★셋 다 `canary_request_lock` 이 **실제로** 건다. 「계산용 dict」가 아니다 — #: SDK 는 지어진 client 에서 **읽어서** 확인한다 #: (`assert_client_retries_observed`). LOCKED_CONTRACT = {"num_retries": 0, "enable_fallback": False, "sdk_max_retries": 0} #: 부트스트랩 **전용** 상한. ★pipeline 것과 섞지 않는다. BOOTSTRAP_CAP = 4 #: ★★★**사람이 승인한** pipeline 비상 상한 (Codex 2026-08-31). #: 계산값이 아니라 **승인값**이 문이다 — 계산이 바뀌어도 승인 없이 안 넓힌다. #: 계산값이 이보다 크면 승인 밖이므로 **선다**. #: ★모드마다 **따로 승인된다** — legacy 판의 승인을 새 판이 물려받지 않는다. #: legacy 92/276 (2026-08-31) · v2_chunk 70/70 (2026-09-02, 시대 원고 4씬). #: ★v2_chunk 는 **93/93 + 검색 10 · 받기 40 · 이미지 0** (Codex 2026-09-02). #: 93 은 여유가 아니라 파생이다 — 이미 쓴 15 + 남은 일반 스텝 38 + #: 중앙 5대상 × 2라운드 × (저작+판정) × 슬롯 2 = 40. #: tier 1 · router 0 · SDK 0 이라 이 범위에서 counted = raw 다. APPROVED_BY_MODE = { "legacy": {"counted": 92, "raw": 276, "search": 0, "download": 0}, # ★2026-09-02 1단계(배경 off · world_guide 까지 글만): 93 → 110 = # 누계 89 + 정상 예상 15 + failover 여유 6. **완주 보장 상한이 아니라** # 정상 15 에 운반 6 을 더 허용한 **정지선**이다 — 넘으면 inconclusive. # 검색 10·받기 40 은 표의 값이고, 중앙 스텝이 이미 끝난 재개 판에서는 # `outbound_doors_on_resume` 가 그 attempt 의 문을 **0** 으로 연다. # ★2026-09-02 재산정 ②: 110 → 120 = 누계 89 + 정상 예상 18(entity_t2i 의 # partial 잔여 3 + 남은 스텝 15) + 여유 13. 슬롯 최악 50 아래의 **정지선**. # ★2026-09-02 재산정 ③ (Codex 조건부 승인): 120 → 135 = closure 정상 합 134 반올림. # 누계 94 → 이번 판 최대 41 · 현실 예상 19~21 · 닿으면 inconclusive. # ★2026-09-02 재산정 ④ (Codex 권고): 135 → 150 = 누계 124 + 정상 11 + 여유 15. # 직전 scene_detail 실패 경로가 18전송이었다 — 145 면 여유 2 뿐. 닿으면 inconclusive. # ★2026-09-02 2단계 (사용자 지시 · Codex 전달): 150 → 210 = 누계 135 + 정상 34 + 여유 41 # (hard 68 까지 총 203 도 안). 사용자가 **완주 속도 우선**을 명시 — 소액 cap 재승인 # 왕복 금지. 이미지는 `APPROVED_IMAGES_BY_SCOPE`(정확한 범위) 에 80. # ★2026-09-02 재산정 ⑥: 210 → 240. still 마다 번역 1 + readback 2 를 계측에 더하자 # closure 상한 정상 합이 237 이 됐다(누계 169 · 실제 남은 것은 S1_Shot2 하나 ≈ 4). # 사용자 지시(속도 우선 · 소액 재승인 왕복 금지)대로 정지선만 올린다. # ★검색 10·받기 40 은 대상 7개 중 **둘**이 다 썼다(실측: O03 은 질의 0 · 문 앞 거절). # 대상 수 × 라운드 × 묶음(≤5) ≈ 70 → 80 · 받기 그 두 배. # ★2026-09-02 밤: 중앙 조사 상한을 production 의무 수(21)로 재니 정상 예상 합 301 — 사용자 지시 # (「상한 승인으로 서지 마라 · 시간이 문제」)대로 글 정지선을 320 으로. 이미지 80 은 그대로. # ★2026-09-03 새벽: 조사가 열여섯 attempt 에 걸쳐 290 을 썼고 이미지 단계의 글(참조 20 · 합성 16 · 씬 1+)이 # 30 으로 모자라 380 으로. 이미지 80 그대로. Codex 에 알림. # ★2026-09-03 03:30 재산정: 380 → 480 = 누계 290 + 5단계 조사 재실행 ≈85(17 대상 × 조사 1 · # 검색 ≤2 · 판정 ≤2) + 이미지 단계 글 ≈37 + 여유 ≈68. 검색 80 → 120 · 받기 160 → 240 도 같은 # 까닭(앞 판이 검색 34 · 받기 ≈136 을 썼고 재실행이 그만큼 더 쓴다). 완주 보장이 아니라 # run 전체 정지선이다 — 닿으면 inconclusive. 사용자: 「시간이 문제, 돈 아님」. "v2_chunk": {"counted": 480, "raw": 480, "search": 120, "download": 240}, } APPROVED_EMERGENCY_COUNTED = APPROVED_BY_MODE["legacy"]["counted"] APPROVED_EMERGENCY_RAW = APPROVED_BY_MODE["legacy"]["raw"] def approved_for(mode: str) -> Dict[str, int]: """그 모드의 **승인 정지선**. ★없는 모드면 선다 — 물려받지 않는다.""" got = APPROVED_BY_MODE.get(str(mode)) if got is None: raise KeyError( f"모드 {mode!r} 의 승인 정지선이 없다 — 사람이 정해야 한다. " f"있는 것: {sorted(APPROVED_BY_MODE)}") return dict(got) #: ★★★**이미지 승인은 0 이다** (실측 2026-09-01). #: 글 예산은 이미지 문을 **못 본다** — `ResearchCallBudget` 은 #: `llm_client._completion` 과 `openai_keys` 세 자리에만 걸려 있고, 유료 #: 이미지는 그 셋을 하나도 안 지난다. 그런데 `scene_detail` 까지의 closure #: 안에 `floor_plan_render` 가 있고 그것은 `gpt-image-2` 를 산다. 글 장부에는 #: **0 으로 적히고** 실제로는 돈이 나가던 자리다. #: ★이 판의 승인 범위는 **글 canary** 다. 이미지는 사람이 따로 정한다 — #: 0 이면 문 앞에서 서고, 어디서 사려 했는지가 장부에 남는다. APPROVED_IMAGE_CALLS = 0 #: ★★이미지 승인은 **정확한 시나리오**(mode·fixture·target·배경·still_recipe·outdoor)에 #: 결속된다 (Codex BLOCK 2026-09-02). 전역 상수 하나를 0→40 으로 바꾸면 그것은 #: 「이번 stage2a 승인」이 아니라 다른 canary 범위에도 열린 값이 된다. 여기 없는 #: 범위는 전부 `APPROVED_IMAGE_CALLS`(0) — 문이 닫힌 채 선다. 전이 기록·재개 문· #: scope 문·dry·run_pipeline 이 **같은 함수**(`approved_images_for`)를 본다. #: 키 = `scope_key(sc)`. 사람이 승인하면 그 표와 **원자적으로** 여기 한 줄을 더한다. APPROVED_IMAGES_BY_SCOPE: Dict[tuple, int] = { # ★2026-09-02 stage2a (사용자 지시 · Codex 전달): 정상 31 · hard 253 은 안 연다. # 80 = 정상의 두 배 재시도(62) + 여유 18. 닿으면 inconclusive · 자동 확대 0. # A/B arm B 는 별도 run_id · 별도 줄이다 — 이 줄을 이어 쓰지 않는다. ("v2_chunk", "period_episode", "scene_image_pipeline", "off", "off", "off"): 80, # ★2026-09-02 현대 배경 축(사용자 지시: A/B 판정 기다리지 말고 병렬 · 넉넉한 상한) ("v2_chunk", "modern_episode", "scene_image_pipeline", "off", "off", "off"): 80, # ★2026-09-03 06:10 통합 canary (Codex 조건 · 사용자 「시간 우선·계속·ASAP」 · goal 「비용이 필요하면 진행」): 배경 on + outdoor on → # scene_detail. 이미지는 floor_plan_render 뿐 — 정상 6 · 재시도 포함 run-wide 18. 최종 씬 이미지 0(scene_image_pipeline 은 닫힘 밖). ("v2_chunk", "period_episode", "scene_detail", "on", "off", "on"): 18, # ★2026-09-03 08:50 한 run 야외 통합 — 같은 run 이라 누계(floor plan 1)에 seed 롤 2 × 묶음 ≤2 × hard 3 = 12 를 더해 run-wide 18 그대로 ("v2_chunk", "period_episode", "outdoor_structure_seed", "on", "off", "on"): 18, } #: ★이 범위에서 이미지를 살 수 있는 스텝 — 계획에 다른 이미지 스텝이 보이면 provider 전에 선다 (Codex 2026-09-03 06:10). #: 없는 범위는 「이미지 스텝 0」이 계약이다(승인 0 과 같다). ALLOWED_IMAGE_STEPS_BY_SCOPE: Dict[tuple, frozenset] = { ("v2_chunk", "period_episode", "scene_image_pipeline", "off", "off", "off"): frozenset({"ref_image_gen", "composite_image_gen", "character_state_variant", "scene_image_pipeline"}), ("v2_chunk", "modern_episode", "scene_image_pipeline", "off", "off", "off"): frozenset({"ref_image_gen", "composite_image_gen", "character_state_variant", "scene_image_pipeline"}), ("v2_chunk", "period_episode", "scene_detail", "on", "off", "on"): frozenset({"floor_plan_render"}), # ★2026-09-03 08:50 한 run 야외 통합(같은 run f7cc45c576c0 · target 만 seed 로): 새로 사는 이미지는 seed 롤뿐 ("v2_chunk", "period_episode", "outdoor_structure_seed", "on", "off", "on"): frozenset({"outdoor_structure_seed"}), } def image_steps_in_plan(built: Dict[str, Any]) -> List[str]: """계획에서 이미지를 살 자리(units_cap > 0 · normal_per_unit > 0)에 있는 스텝들.""" return sorted(r["step"] for r in ((built.get("images") or {}).get("rows") or []) if int(r.get("units_cap") or 0) > 0 and int(r.get("normal_per_unit") or 0) > 0) def assert_image_sources(built: Dict[str, Any], sc: Dict[str, Any]) -> Dict[str, Any]: """이미지를 사는 스텝이 이 범위의 허용 집합 안에만 있나 — 밖의 것이 보이면 provider 전에 선다. ★허용표에 **없는** 범위는 적기만 한다 — 그 범위는 이미지 승인 0 이라 live 가 제 문(승인 대조)에서 제 문구로 선다; 여기서 먼저 서면 그 문·문구(router 잠금 등)를 가린다(실측: 다른 범위 시험 둘이 이 문구로 섰다).""" got = set(image_steps_in_plan(built)) key = scope_key(sc) allowed = set(ALLOWED_IMAGE_STEPS_BY_SCOPE.get(key, frozenset())) extra = sorted(got - allowed) if extra and key in ALLOWED_IMAGE_STEPS_BY_SCOPE: raise ScopeMismatch(f"이 범위에서 허용되지 않은 이미지 스텝이 계획에 있다 {extra} — 허용 {sorted(allowed)}. 사지 않고 선다") return {"image_steps": sorted(got), "allowed": sorted(allowed), "outside": extra} def scope_key(sc: Dict[str, Any]) -> tuple: """이미지 승인의 열쇠 — 시나리오 축 전부, 순서 고정.""" return tuple(str(sc.get(k)) for k in _SCENARIO_AXES) def approved_images_for(sc: Dict[str, Any]) -> int: """이 정확한 범위에 사람이 승인한 이미지 run 전체 상한. 없으면 0.""" return int(APPROVED_IMAGES_BY_SCOPE.get(scope_key(sc), APPROVED_IMAGE_CALLS)) class ScopeMismatch(RuntimeError): """도는 것이 **승인한 것과 다르다**. ★provider 0 으로 선다.""" def scenario(*, mode: str = DEFAULT_CANARY_MODE, fixture: str = cbs.DEFAULT_FIXTURE, target: str = "scene_detail", background: Optional[str] = None, still_recipe: Optional[str] = None, outdoor: Optional[str] = None) -> Dict[str, Any]: """이 판이 **무엇을 도는가** — 한 값. ★사슬 전체가 이것만 본다. ★★★세 축이 흩어져 있으면 하나가 기본값으로 조용히 떨어진다 (Codex BLOCK 2026-09-02). 실제로 그랬다 — `fixture_config(mode)`· `fixture_dimensions(fixture)`·`manuscript_pdf(fixture=)`· `approved_for(mode)` 를 다 만들어 놓고 `run()` 이 **하나도 안 넘겨서**, 유료로 열면 legacy·canary_one_scene·scene_detail·92/276 이 돌 판이었다. Raises: ScopeMismatch: 모르는 모드·원고·목표. ★기본값으로 안 떨어진다. """ from app.core.grounding_mode import GROUNDING_MODES from app.core.step_manifest import STEP_MANIFEST as M if mode not in GROUNDING_MODES: raise ScopeMismatch(f"모르는 모드 {mode!r} — {sorted(GROUNDING_MODES)}") if fixture not in cbs.FIXTURES: raise ScopeMismatch(f"모르는 원고 {fixture!r} — {cbs.FIXTURES}") if target not in M: raise ScopeMismatch(f"모르는 목표 {target!r}") declared = {"background": background, "still_recipe": still_recipe, "outdoor": outdoor} axes: Dict[str, str] = {} for axis in _SETTING_AXES: actual = axis_actual(axis) want = declared.get(axis) if want is not None and want != actual: # ★설정은 process 가 뜰 때 고정된다(`settings`). 안에서 못 바꾸므로 # 선언과 다르면 **선다** — env 를 process 앞에 준다. raise ScopeMismatch( f"{axis} 선언은 {want!r} 인데 이 process 는 {actual!r} 로 떴다 — " f"`{_SETTING_AXES[axis]['env']}` 를 process 시작 전에 줘라") axes[axis] = actual sc = {"mode": mode, "fixture": fixture, "target": target, **axes, "approved": approved_for(mode)} sc["approved_images"] = approved_images_for(sc) # ★이 범위에만 결속된 값 return sc #: 아무 인자도 안 주면 도는 판. ★옛 legacy canary 그대로다. DEFAULT_SCENARIO = None # ★`_scenario()` 가 늦게 만든다 def _scenario(sc: Optional[Dict[str, Any]]) -> Dict[str, Any]: """★`None` 이면 기본이되, **무엇이 기본인지 산출에 적힌다**.""" return sc if sc is not None else scenario() def git_tip() -> Dict[str, Any]: """지금 도는 **코드의 신원**. ★더러운 트리면 그 사실을 적는다.""" def _sh(*args, raw: bool = False): out = subprocess.run(args, cwd=str(BACKEND.parent), capture_output=True, text=True).stdout return out if raw else out.strip() # ★`strip()` 을 걸면 안 된다 — porcelain 은 앞 두 칸이 **상태 칸**이라 # ' M path' 의 첫 칸이 잘리고 그러면 `[3:]` 이 경로 첫 글자를 먹는다 # (실측: 'ackend/tools/...'). 막는 문이 **없는 경로**를 대는 꼴이다. dirty = _sh("git", "status", "--porcelain", raw=True) return {"tip": _sh("git", "rev-parse", "HEAD"), "clean": not dirty.strip(), "dirty_files": [x[3:] for x in dirty.splitlines() if x][:20]} def fixture_dimensions(fixture: str = cbs.DEFAULT_FIXTURE) -> Dict[str, int]: """이 원고의 **치수**. ★상한을 여기서 뽑는다 — 고정값을 안 박는다. ★앞 판은 fan_out 스텝을 **전부 2**로 셌다(1씬 2샷 때의 수). 원고가 2씬 3샷으로 늘었는데 그대로 두면 `expected` 가 **옛 판의 수**가 된다 (Codex 2026-09-01). """ sys.path.insert(0, str(BACKEND / "tests")) fx = cbs.load_fixture(fixture) if hasattr(fx, "assert_shape"): fx.assert_shape() scenes = len(fx.segments()) shots = sum(len(s["shots"]) for s in fx.shot_scenes()) return {"scenes": scenes, "shots": shots, # ★fan_out 의 **단위**는 스텝마다 다르다(씬·샷). 둘 중 큰 것을 # 쓴다 — 작은 것을 쓰면 상한이 실제보다 낮아 헛되이 막는다. "fan_out_cap": max(scenes, shots), # ★엔티티 단위 스텝의 상한은 원고가 **선언**한다 (Codex 재리뷰 # 2026-09-02: max(씬,샷)=10 으로 접었는데 실제 엔티티는 15 였다). # 선언이 없으면 None — entity 단위 스텝이 살 자리면 선다. "entity_cap": (int(getattr(fx, "ENTITY_CAP")) if hasattr(fx, "ENTITY_CAP") else None), # ★이미지 스텝 단위의 상한 — 없으면 None(그 단위 스텝이 살 자리면 선다) "outlook_pair_cap": (int(getattr(fx, "OUTLOOK_PAIR_CAP")) if hasattr(fx, "OUTLOOK_PAIR_CAP") else None), "state_variant_cap": (int(getattr(fx, "STATE_VARIANT_CAP")) if hasattr(fx, "STATE_VARIANT_CAP") else None), # ★야외 건물 묶음 상한 — 원고가 선언한다(실외 장소 수 기준). 없으면 야외 단위 스텝이 살 자리면 선다 "outdoor_group_cap": (int(getattr(fx, "OUTDOOR_GROUP_CAP")) if hasattr(fx, "OUTDOOR_GROUP_CAP") else None), # ★배경 사슬 세 단위 — 원고가 선언한 상한(production 산출과 대조하는 문이 있다). 없으면 그 단위 스텝이 살 자리면 선다 "chain_bg_group_cap": (int(getattr(fx, "CHAIN_GROUP_CAP")) if hasattr(fx, "CHAIN_GROUP_CAP") else None), "floor_plan_cap": (int(getattr(fx, "FLOOR_PLAN_CAP")) if hasattr(fx, "FLOOR_PLAN_CAP") else None), "background_cap": (int(getattr(fx, "BACKGROUND_CAP")) if hasattr(fx, "BACKGROUND_CAP") else None)} #: ★★★**production 호출 구조**가 씬 단위인 스텝 (Codex BLOCK 2026-09-02). #: manifest 의 `fan_out` 만 보면 틀린다 — `shot_director` 는 `fan_out=False` #: 인데 `direct_shots()` 가 **씬마다** `_resolve_scene_llm` 을 부르고, #: `shot_selection` 은 `fan_out=True` 지만 단위가 샷이 아니라 **씬**이다. #: 앞 판은 전자를 1, 후자를 샷 수(10)로 셌다 → 전자는 세 번째 씬에서 #: **정상 경로에서도 반드시 거절**되고, 후자는 과대였다. #: ★이 표가 **수와 문의 같은 SOT** 다 — `build_plan` 의 `caps` 가 이것을 #: 쓰고, `canary_pipeline` 의 스텝 문은 그 `caps` 를 쓴다. #: ★손으로 적은 목록이므로 **production 코드와 대조하는 시험**이 잠근다 #: (`test_the_cap_follows_the_production_fan_out_unit`) — 씬 loop 안에 #: LLM 호출이 있는지 AST 로 본다. #: ★★★**계측 단위의 정본은 명시 계약**이다 (Codex 재리뷰 2026-09-02) — manifest #: `fan_out` 도, AST 자동추론도 정본이 아니다. 실측 셋: `shot_director`(아침)· #: `scene_camera_flow`(오후 · attempt 0dfb50f3f95b · 유료 5)는 `fan_out=False` 인데 #: 씬마다 부르고, `shot_staging` 은 스텝 파일엔 호출이 없지만 helper #: (`run_shot_staging`)가 선택 샷을 `BATCH_SIZE` 로 나눈 batch 마다 부른다. #: **읽어서 확인한 것만** 단위를 적고, 나머지는 `unverified` 로 둔다 — #: unverified 스텝이 **살 자리**(끝나지 않은 채 closure 안)에 있으면 live 는 #: provider 앞에서 선다. single 로 접지 않는다. UNIT_SINGLE = "single" # 판마다 한 번 UNIT_SCENE = "scene" # 씬마다 한 번 UNIT_SHOT = "shot" # 샷(또는 묶음)마다 — fixture fan_out_cap 상한 UNIT_ENTITY = "entity" # 엔티티마다 — fixture fan_out_cap 상한 UNIT_GROUP = "group" # 데이터가 정한 묶음마다 — production 상한(group_cap_of) UNIT_BATCH = "selected_shot_batch" # 선택 샷을 BATCH_SIZE 로 나눈 batch 마다 UNIT_CENTRAL = "central" # 중앙 조사 — `central_logical_cap` 이 따로 센다 UNIT_UNVERIFIED = "unverified" # 아직 안 읽었다 — 살 자리면 선다 UNIT_OUTLOOK_PAIR = "outlook_pair" # (인물, 아웃룩) 쌍마다 — fixture OUTLOOK_PAIR_CAP 상한 UNIT_STATE_VARIANT = "state_variant" # shot_staging 이 정한 (인물, 상태) 마다 — fixture STATE_VARIANT_CAP 상한 #: ★2026-09-03 (Codex · 야외 canary 최소 closure): 야외 **건물 묶음**(background_classify 의 building_group 중 #: 실외 멤버가 있는 것)마다 — fixture OUTDOOR_GROUP_CAP 상한. 읽어서 확인한 loop: #: outdoor_place_spec_step:186 `for g, outdoor_members in outdoor_groups` → author_place_spec(while attempts < MAX_ATTEMPTS) #: outdoor_lane_plan_step:170 `for gid in sorted(target)` → plan(while attempts < max_attempts=3) #: outdoor_structure_form_reference(v2_chunk · 앞쪽 소유): 보충 대상(≤ 묶음)마다 ca.run — 조사 1 + 검색 ≤2 + 판정 ≤2 UNIT_OUTDOOR_GROUP = "outdoor_group" #: ★배경 사슬 네 스텝 (2026-09-03 05:30 loop 읽어서 확인) — fixture 상한 + production 산출과 대조하는 provider 앞 문: #: background_master_plan_step:223 ThreadPool(chain_groups) · 그룹마다 call_structured 1 → chain_bg 그룹마다 #: floor_plan_prompt_step:243 fp_jobs(plan.floor_plans) 마다 call_structured 1 → floor plan 마다 #: floor_plan_render_step:488 renderable fp 마다 이미지 1 (SDK 기본 재시도 2 → hard 3) → floor plan 마다(이미지 문) #: background_prompt_step:587 level 의 bid(plan.backgrounds) 마다 call_structured 1 → background 마다 UNIT_CHAIN_GROUP = "chain_bg_group" # fixture CHAIN_GROUP_CAP 상한 · 술어 background_master_plan_step.chain_groups_of UNIT_FLOOR_PLAN = "floor_plan" # fixture FLOOR_PLAN_CAP 상한 · background_master_plan.floor_plans_of UNIT_BACKGROUND = "background" # fixture BACKGROUND_CAP 상한 · background_master_plan.backgrounds_of METERING_UNITS: Dict[str, str] = { # 읽어서 확인한 것 (파일:함수) "shot_selection": UNIT_SCENE, # shot_selection_step: pool.submit(_select_for_scene) per scene "shot_director": UNIT_SCENE, # pipeline/shot_director.direct_shots: _resolve_scene_llm per scene "scene_camera_flow": UNIT_SCENE, # scene_camera_flow_step: task per scene → _process_scene "scene_consistency": UNIT_SCENE, # scene_consistency_step: task per scene → _process_one_scene "scene_detail": UNIT_SHOT, # detail_steps: _run_shots_with_retry per shot "entity_t2i": UNIT_ENTITY, # entity_steps: worker per entity (남은 것만 · partial 재개) "visual_continuity_anchor": UNIT_GROUP, # anchor_provider per prop group "shot_staging": UNIT_BATCH, # pipeline/shot_staging.run_shot_staging: batch of BATCH_SIZE "world_guide": UNIT_SINGLE, # world_guide_generator.generate: call_structured 한 번 "reference_acquisition": UNIT_CENTRAL, # 아직 안 읽은 것 — 이 run 에서는 전부 끝났거나 closure 밖이다 "text_cleanup": UNIT_SINGLE, "scene_segmentation": UNIT_SINGLE, # ★2026-09-02: fixture 선언(has_outlooks·outdoor_direct_or_map_or_lane· # background_share_plan)으로 closure 에 들어올 수 있는 셋 — 아직 안 읽었다. # lane 을 끈 canary 에선 건너뛰고, 켠 판이면 살 자리라 live 가 선다. "outdoor_place_spec": UNIT_OUTDOOR_GROUP, # outdoor_place_spec_step:186 실외 묶음마다 author_place_spec "outdoor_place_canon": UNIT_UNVERIFIED, "background_share_plan": UNIT_UNVERIFIED, # ★이미지 스텝 넷 — 2026-09-02 읽어서 확인. 글 문은 dual LVM(ask_both = 2호출), # 이미지 문은 IMAGE_CALLS_PER_UNIT 이 따로 센다. "composite_image_gen": UNIT_OUTLOOK_PAIR, # reference_phase2_service:102 아웃룩마다 + phase3:86-90 (인물,아웃룩) 쌍마다 — 쌍 단위(phase2 ≤ 쌍 수) "episode_summary": UNIT_SINGLE, "visual_world_rules": UNIT_SINGLE, "entity_character_list": UNIT_SINGLE, "scene_summary": UNIT_SCENE, "beat_extract": UNIT_SCENE, "shot_extract": UNIT_SCENE, "shot_validator": UNIT_SCENE, "grounding_chunk": UNIT_SCENE, "entity_all_character": UNIT_UNVERIFIED, "entity_extract_character": UNIT_UNVERIFIED, "entity_all_location": UNIT_UNVERIFIED, "entity_extract_location": UNIT_UNVERIFIED, "entity_all_prop": UNIT_UNVERIFIED, "entity_extract_prop": UNIT_UNVERIFIED, "entity_merge": UNIT_SINGLE, "entity_relation": UNIT_SINGLE, "entity_filter": UNIT_SINGLE, "entity_detail": UNIT_SINGLE, "scene_director": UNIT_SINGLE, "outlook_phase1": UNIT_SINGLE, "outlook_phase2": UNIT_SINGLE, "outlook_phase3": UNIT_SINGLE, "background_classify": UNIT_SINGLE, # background_classify.run_background_classify: call_structured_fn 한 번(재시도 loop 안) "background_master_plan": UNIT_CHAIN_GROUP, # background_master_plan_step:223 chain_groups 마다 _process "floor_plan_prompt": UNIT_FLOOR_PLAN, # floor_plan_prompt_step:243 fp_jobs 마다 _process "floor_plan_render": UNIT_FLOOR_PLAN, # floor_plan_render_step:488 renderable fp 마다 이미지 (이미지 문) "background_prompt": UNIT_BACKGROUND, # background_prompt_step:587 level 의 bid 마다 _process "outdoor_lane_plan": UNIT_OUTDOOR_GROUP, # outdoor_lane_plan_step:170 대상 묶음마다 plan # ★2026-09-03 08:50 (한 run 야외 통합 · loop 읽어서 확인): shot_ref_classify.run_shot_ref_classify 는 선택 샷 **전부를 한 글**로 # 넘겨 판마다 고정 5회(bg·prev·tod·place·world_anchor — :245/:269/:286/:319/:396) · `_call_items_with_retry` max_attempts=2 "shot_ref_classify": UNIT_SINGLE, "shot_continuity": UNIT_UNVERIFIED, "outdoor_structure_form_reference": UNIT_OUTDOOR_GROUP, # 앞쪽 소유 판: 보충 대상(≤ 묶음)마다 ca.run # ★2026-09-03 08:50 읽어서 확인: outdoor_structure_seed_step:1226 `for gid in target_gids`(lane plan 의 structure_plate 묶음 ⊆ 실외 묶음) # → run_multiroll_select: 롤 이미지 still_recipe_roll_count(기본 2, :1085) + 판정 1(multiroll_select:1743 · 이중 심판 = 물리 2) # · critique/fix 는 still_recipe_critique_enabled(기본 False) · 변형 저작·스케치는 structure_seed_variants_enabled(기본 False) "outdoor_structure_seed": UNIT_OUTDOOR_GROUP, "ref_image_gen": UNIT_ENTITY, # reference_phase1_service:88-93 batch 의 엔티티마다 _gen_ref (canonical 갈래 C/P · 있는 것은 already_done) "character_state_variant": UNIT_STATE_VARIANT, # image_steps CharacterStateVariantStep: shot_staging character_angles 의 (인물, 상태) 마다 generate_and_validate_reference "background_render": UNIT_UNVERIFIED, "shot_conti_light": UNIT_UNVERIFIED, "scene_image_pipeline": UNIT_SHOT, # image_steps:1826 is_selected still 마다 → coordinator:2043 variation = settings.scene_variation_count } #: 파생 — 옛 이름을 쓰는 시험·호출자용 SCENE_UNIT_STEPS: tuple = tuple(k for k, v in METERING_UNITS.items() if v == UNIT_SCENE) GROUP_UNIT_STEPS: tuple = tuple(k for k, v in METERING_UNITS.items() if v == UNIT_GROUP) def metering_unit_of(step: str) -> str: """그 스텝의 계측 단위. ★표에 없으면 **선다** — 모르는 채 안 센다.""" if step not in METERING_UNITS: raise ScopeMismatch( f"계측 단위 계약에 {step!r} 가 없다 — single 로 접지 않는다. " "읽어서 확인한 뒤 METERING_UNITS 에 적어라") return METERING_UNITS[step] def shot_staging_batch_size() -> int: """`run_shot_staging` 의 BATCH_SIZE — **production 파일에서 정적으로** 읽는다 (import 하면 llm_client 가 딸려 온다).""" return _static_int_constant("app/modules/pipeline/shot_staging.py", "BATCH_SIZE") def group_cap_of(step: str) -> int: """`group` 단위 스텝의 **production 상한** — fixture 치수가 아니라 그 스텝이 실제로 적용하는 설정값에서 읽는다 (Codex 재리뷰 2026-09-02). `visual_continuity_anchor` 는 두 갈래를 **독립된 상한**으로 산다 — prop 묶음 `visual_continuity_anchor_group_cap`(기본 8) + 고정 인물 묶음 `immobilized_subject_continuity_cap`(기본 4 · 그 갈래가 켜져 있을 때). fixture fan_out_cap(10)으로 접으면 최대 12 를 10 으로 **낮게** 센다. ★새 group 스텝은 여기 갈래를 **적어야** 센다 — 모르면 선다. """ from app.core.config import settings if step == "visual_continuity_anchor": cap = int(getattr(settings, "visual_continuity_anchor_group_cap", 0) or 0) if bool(getattr(settings, "immobilized_subject_continuity_enabled", False)): cap += int(getattr(settings, "immobilized_subject_continuity_cap", 0) or 0) return max(cap, 0) raise ScopeMismatch( f"{step!r} 의 group 상한을 어디서 읽는지 안 적었다 — 모르는 채 안 센다") def fan_out_unit_of(step: str) -> str: """옛 이름 — 단위를 `scenes`·`fan_out_cap`·`batch`·`one` 으로 접는다. ★`unverified` 는 manifest 로 접되 **표시**된다(`unverified_units_in`).""" from app.core.step_manifest import STEP_MANIFEST as M u = metering_unit_of(step) if u == UNIT_SCENE: return "scenes" if u == UNIT_SHOT: return "fan_out_cap" if u == UNIT_ENTITY: return "entity" if u == UNIT_GROUP: return "group" if u == UNIT_BATCH: return "batch" if u == UNIT_OUTLOOK_PAIR: return "outlook_pair" if u == UNIT_STATE_VARIANT: return "state_variant" if u == UNIT_OUTDOOR_GROUP: return "outdoor_group" if u in (UNIT_CHAIN_GROUP, UNIT_FLOOR_PLAN, UNIT_BACKGROUND): return u if u in (UNIT_SINGLE, UNIT_CENTRAL): return "one" return "fan_out_cap" if M.get(step, {}).get("fan_out") else "one" def logical_cap_of(step: str, dims: Dict[str, int]) -> int: """그 스텝의 **논리 상한** — 단위 하나에서 나온다. ★두 자리에 안 적는다.""" import math unit = fan_out_unit_of(step) if unit == "scenes": return int(dims["scenes"]) if unit == "fan_out_cap": return int(dims["fan_out_cap"]) if unit == "batch": return max(1, math.ceil(int(dims["shots"]) / shot_staging_batch_size())) if unit == "group": return group_cap_of(step) if unit == "entity": cap = dims.get("entity_cap") if not cap: raise ScopeMismatch( f"{step!r} 는 엔티티 단위인데 원고가 ENTITY_CAP 을 선언하지 않았다 — " "max(씬,샷)으로 접지 않는다") return int(cap) if unit in ("outlook_pair", "state_variant", "outdoor_group", "chain_bg_group", "floor_plan", "background"): key = f"{unit}_cap" cap = dims.get(key) if cap is None: raise ScopeMismatch( f"{step!r} 는 {unit} 단위인데 원고가 {key.upper()} 을 선언하지 않았다 — " "접지 않는다") return int(cap) return 1 #: ★★단위 **안에서** production 이 허용하는 유료 호출 횟수 (Codex 재리뷰 2026-09-02). #: 단위 수만 정본화하면 그 안의 재시도가 문 앞에서 막혀 스텝이 실패한다 — #: shot_staging 은 batch 마다 MAX_ATTEMPTS 안에서 부르고, scene_detail 은 샷마다 #: 기본 1 + 조건부 수정 재호출 1(`if to_remove or fsc_echo_violations`) + #: variation(≤2) 마다 [owned judge 1 + redraw 면 `_attempt_owned_redraw_repair`: #: repair 시도 ≤2(모델 둘) + verify judge 1 + 상위 모델 재시도(repair 1 + verify 1) #: = 5] = 6 → 1 + 1 + 2×6 = 14, 실패 샷은 `_run_shots_with_retry` 가 한 번 더 #: `_analyze_one` 한다 → **28**. 정상 예상은 단위당 1 이고 이것은 **hard cap** 이다 #: — 둘을 갈라 적는다. #: ★값을 손으로 적은 것은 시험이 production 상수·구조와 대조한다. CALLS_PER_UNIT: Dict[str, Any] = { # ★2026-09-02 앞단 글 스텝 18 — 2026-09-02 읽어서 확인 + period run(4씬·10샷) 실측 counted 로 대조: # text_cleanup 1 · scene_segmentation 1(segment_rule_author) · episode_summary 1 · visual_world_rules 1 # · entity_character_list 1(character_list_step:74 max_retry 2 → 3) · scene_summary 씬마다 1(실측 4) # · beat/shot_extract 번들(≤씬)마다 1 · beat_shot_steps:232 max_retry 5 → 6 · shot_validator 씬마다 1(실측 4) # · grounding_chunk 구간(≤씬)마다 1(+재시도 여유 1) · entity_merge/relation/filter 는 v2_chunk 에서 투영(실측 0 · 상한 1) # · entity_detail 실측 2(entity_extractor_v3 range(3)) → 4 · scene_director 1 · outlook_phase1/2/3 실측 1 → 2. "entity_character_list": 3, "beat_extract": 6, "shot_extract": 6, "grounding_chunk": 2, "entity_detail": 4, "outlook_phase1": 2, "outlook_phase2": 2, "outlook_phase3": 2, "shot_staging": "MAX_ATTEMPTS", # pipeline/shot_staging.py 상수 — 파일에서 읽는다 # ★야외 최소 closure (2026-09-03) — 읽어서 확인: "background_classify": 3, # background_classify.py:186 max_retries=3 (한 호출 · 재시도 loop) "outdoor_place_spec": ("app/modules/pipeline/outdoor_place_spec.py", "MAX_ATTEMPTS"), # :296 while attempts < MAX_ATTEMPTS "outdoor_lane_plan": 3, # outdoor_lane_plan.py:509 max_attempts=3 # ★배경 사슬 (2026-09-03 05:30 읽어서 확인): 단위마다 call_structured 1 — canary 는 num_retries 0 으로 잠그므로 여유 1 → 2 "background_master_plan": 2, "floor_plan_prompt": 2, "background_prompt": 2, # ★실측 f7cc45c576c0 4판 (2026-09-03 07:18): 운반층은 images.generate 도 **스텝 문**(research_call_budget · step cap)에 예약한다 — # 0 으로 두면 run-wide 이미지 문이 열려 있어도 안쪽 문에서 전부 거절된다. floor plan 마다 이미지 1 · floor_plan_render.py:42 max_attempts=3. "floor_plan_render": ("app/modules/pipeline/floor_plan_render.py", "MAX_ATTEMPTS") if False else 3, "outdoor_structure_form_reference": 5, # 보충 대상마다 조사 1 + 검색 2 + 판정 2 (ca.run · 2라운드) # ★2026-09-03 08:50 한 run 야외 통합: 판마다 5 호출 × max_attempts 2 → 8(tod·world_anchor 는 재시도 없음) "shot_ref_classify": 8, # ★묶음마다 이미지 롤 2 × SDK 재시도(hard 3) = 6 + 판정 물리 2(+여유 2) = 10 — 이미지도 스텝 문에 예약된다(floor_plan_render 실측) "outdoor_structure_seed": 10, "entity_t2i": 3, # entity_steps: `for attempt in range(3)` "scene_detail": 28, # 위 주석 — (1 + 1 + 2×6) × 2 # ★이미지 스텝 넷의 **글 문**(LVM) — 2026-09-02 읽어서 확인: # ref_image_pipeline.generate_and_validate_reference: 검증 `_call_gpt_lvm` → # `dual_vlm.ask_both`(GPT+Gemini = **2호출**) 1회 + severity=severe 면 재생성 뒤 # 비교 `ask_both` 1회(2호출) → 정상 2 · hard 4. "ref_image_gen": 4, "composite_image_gen": 8, # phase2(아웃룩) 1 생성 + phase3(쌍) 1 생성 = 같은 pipeline × 2 "character_state_variant": 4, # 같은 generate_and_validate_reference # scene_image_pipeline: `_decide_scene_lvm`(settings.scene_lvm_validation_mode) 가 # off 면 글 0. 켜면 variation 마다 검증 2 + severe 비교 2. ★settings 에서 읽는다. "scene_image_pipeline": "scene_lvm", } #: 스텝이 **단위와 무관하게 판마다** 하는 글 호출 — 정상·hard 둘 다에 더한다. #: ★실측 2026-09-02 stage2a attempt 2620121dc61d: `scene_image_pipeline` 은 still 마다가 #: 아니라 **진입 시 한 번** `resolve_world_guide` 가 world guide 를 다시 만든다 #: (scene_persistence_service:844 → world_guide_generator:131 call_structured — 해시가 #: entities·stills 수로 바뀌면 재생성). 내가 0 으로 적어 글 문이 provider 앞에서 세웠다 #: (유료 0 · fail-closed 는 맞게 섰다). 읽은 대로 1. EXTRA_CALLS_PER_RUN: Dict[str, int] = {"scene_image_pipeline": 1} def extra_calls_per_run_of(step: str) -> int: return int(EXTRA_CALLS_PER_RUN.get(step, 0)) #: 단위마다 **한 번씩만 성공했을 때**의 글 호출 — 기본 1. dual LVM 은 성공 한 번이 2호출. NORMAL_CALLS_PER_UNIT: Dict[str, Any] = { "ref_image_gen": 2, "composite_image_gen": 4, "character_state_variant": 2, "outdoor_structure_form_reference": 3, # 보충 대상마다 정상: 조사 1 + 검색 1 + 판정 1 "shot_ref_classify": 5, # 판마다 bg·prev·tod·place·world_anchor 한 번씩 "outdoor_structure_seed": 4, # 묶음마다 이미지 롤 2 + 판정 물리 2 "background_master_plan": 1, "floor_plan_prompt": 1, "background_prompt": 1, "floor_plan_render": 1, "scene_image_pipeline": "scene_lvm_normal", } def _scene_lvm_calls(*, normal: bool) -> int: """씬 이미지의 still 당 **글** 호출 — settings 한 곳에서. ★실측 2026-09-02 stage2a attempt 6b2c5b581d83 (still 당 0 으로 적었다가 문에 걸림): · `prompt_translation`(t2i_prompt_composer:167 · gpt-mini): still 마다 1 — Tier1 + sanitized + GPT fallback 으로 hard 3 · `single_frame_readback`(scene_image_pipeline:309 · settings.scene_single_frame_ readback_enabled): 이미지마다 판정 1(+ 위반이면 correction 재생성 뒤 재판정 1) · scene LVM(`scene_lvm_validation_mode`): off 면 0 · 켜면 variation 마다 2/4 """ from app.core.config import settings n = int(getattr(settings, "scene_variation_count", 0) or 0) translation = 1 if normal else 3 readback = 0 if bool(getattr(settings, "scene_single_frame_readback_enabled", False)): readback = n * (1 if normal else 2) lvm = 0 if str(getattr(settings, "scene_lvm_validation_mode", "off")) != "off": lvm = n * (2 if normal else 4) return translation + readback + lvm def calls_per_unit_of(step: str) -> int: v = CALLS_PER_UNIT.get(step, 1) if v == "scene_lvm": return _scene_lvm_calls(normal=False) if isinstance(v, tuple): return _static_int_constant(v[0], v[1]) # (파일, 상수 이름) — 그 파일에서 읽는다 if isinstance(v, str): return _static_int_constant("app/modules/pipeline/shot_staging.py", v) return int(v) def normal_calls_per_unit_of(step: str) -> int: v = NORMAL_CALLS_PER_UNIT.get(step, 1) if v == "scene_lvm_normal": return _scene_lvm_calls(normal=True) return int(v) #: 이미지 문 — 단위마다 (정상, hard). ★글 문과 **다른 문**이다(`canary_image_scope`). #: 2026-09-02 읽어서 확인: #: ref/composite/state: ref_image_pipeline:341 `for attempt in range(4)`(1 + sanitized 3) #: + :425 severe 재생성 1 → 정상 1 · hard 5. composite 는 phase2+phase3 두 생성. #: scene: coordinator:2043 variation N(settings.scene_variation_count) · :2888 #: `for _attempt in range(3)` · scene_image_pipeline:423 `for attempt in range(4)` #: → 정상 N · hard N×3×4 (LVM off 면 severe 재생성 없음). IMAGE_CALLS_PER_UNIT: Dict[str, Any] = { "ref_image_gen": (1, 5), "composite_image_gen": (2, 10), "character_state_variant": (1, 5), "scene_image_pipeline": "scene_variations", "floor_plan_render": (1, 3), # floor_plan_render_step:86 — SDK 기본 재시도 2 → hard 3 "outdoor_structure_seed": (2, 6), # multiroll_select:1659 롤마다 gen_fn 1 × still_recipe_roll_count 2 · gpt_image_gen 은 loop 없음(SDK 재시도 → hard 3) } def image_calls_per_unit_of(step: str) -> tuple: v = IMAGE_CALLS_PER_UNIT.get(step) if v is None: return (0, 0) if v == "scene_variations": from app.core.config import settings n = int(getattr(settings, "scene_variation_count", 0) or 0) hard = n * 3 * 4 # ★readback 이 켜진 판은 위반 시 correction 재생성이 이미지 1 더 (hard 에만) if bool(getattr(settings, "scene_single_frame_readback_enabled", False)): hard += n return (n, hard) return (int(v[0]), int(v[1])) def image_plan(dims: Dict[str, Any], metered: Sequence[str]) -> Dict[str, Any]: """이미지 문의 계획표 — 스텝마다 단위 상한 × (정상, hard). ★승인은 정상 합과 견준다.""" rows = [] for s in metered: n, h = image_calls_per_unit_of(s) if (n, h) == (0, 0): continue units = logical_cap_of(s, dims) rows.append({"step": s, "unit": metering_unit_of(s), "units_cap": units, "normal_per_unit": n, "hard_per_unit": h, "normal": units * n, "hard": units * h}) return {"rows": rows, "normal_total": int(sum(r["normal"] for r in rows)), "hard_total": int(sum(r["hard"] for r in rows)), "★means": ("정상 = 단위마다 한 번씩 성공 · hard = 재시도가 다 터졌을 때. " "문은 run-wide `canary_image_scope(cap=APPROVED_IMAGE_CALLS)` 하나다")} def _static_int_constant(rel: str, name: str) -> int: """production 파일의 모듈 상수를 **import 없이** 읽는다.""" import ast as _ast src = (BACKEND / rel).read_text(encoding="utf-8") for n in _ast.parse(src).body: if isinstance(n, _ast.Assign) and len(n.targets) == 1 \ and getattr(n.targets[0], "id", "") == name \ and isinstance(n.value, _ast.Constant): return int(n.value.value) raise ScopeMismatch(f"{rel} 의 {name} 을 못 읽었다") def hard_cap_of(step: str, dims: Dict[str, int]) -> int: """스텝 문에 거는 **hard cap** = 단위 수 × 단위당 허용 호출.""" return logical_cap_of(step, dims) * calls_per_unit_of(step) def _last_cp(run_id: str, step: str) -> Optional[Dict[str, Any]]: root = ci.root_dir(run_id) cps = sorted((root / "projects").glob(f"*/checkpoints/episodes/*/{step}/manifest.json")) if not cps: return None return json.loads(cps[-1].read_text(encoding="utf-8")) or {} def expected_units_of(run_id: str) -> Dict[str, int]: """이미지 스텝 넷이 **이 run 에서 실제로 살 단위 수** — production 산출(CP)에서 읽는다. ★상한(fixture 선언)과 다르다: 상한은 문에 걸고, 이 수는 승인표의 「정상」이다 (Codex 2026-09-02: 승인 = normal + 명시 여유). 없으면 0 — 「못 봤다」가 아니라 그 CP 가 아직 없다는 뜻이고, 그 스텝은 아직 살 자리가 아니다. ref_image_gen: entity_detail 큐의 canonical 갈래(C/P) − reference_checkpoint 완료 composite_image_gen: outlook_phase3 의 아웃룩 수(쌍 하나 = 아웃룩 하나) character_state_variant: shot_staging 의 눕는 (인물, 상태) — production 술어 scene_image_pipeline: scene_detail 의 카드 수(선택 still) """ from app.core.subject_state import is_immobilized_state from app.modules.pipeline.grounding_entity_contract import canonical_ref_owner_types out = {"ref_image_gen": 0, "composite_image_gen": 0, "character_state_variant": 0, "scene_image_pipeline": 0} owners = set(canonical_ref_owner_types()) ed = _last_cp(run_id, "entity_detail") if ed: queue = ((ed.get("data") or {}).get("entity_queue")) or [] # ★큐 항목은 [이름, 종류, short_id] 목록 — 종류로만 가른다 targets = [q for q in queue if isinstance(q, (list, tuple)) and len(q) >= 2 and str(q[1]) in owners] done_ids: set = set() root = ci.root_dir(run_id) for f in sorted((root / "projects").glob("*/checkpoints/images/*/reference_checkpoint.json")): done_ids |= set(((json.loads(f.read_text(encoding="utf-8")) or {}).get("completed")) or {}) out["ref_image_gen"] = max(0, len(targets) - len(done_ids)) op = _last_cp(run_id, "outlook_phase3") if op: out["composite_image_gen"] = len(((op.get("data") or {}).get("outlooks")) or []) st = _last_cp(run_id, "shot_staging") if st: pairs = set() for shot in ((st.get("data") or st).get("shots") or []): for ca in (shot.get("character_angles") or []): state = ca.get("subject_state") if state is not None and is_immobilized_state(str(state)): pairs.add((str(ca.get("character") or ""), str(state))) out["character_state_variant"] = len(pairs) sd = _last_cp(run_id, "scene_detail") if sd: out["scene_image_pipeline"] = len(((sd.get("data") or {}).get("scenes")) or []) return out def expected_calls_of(run_id: str) -> Dict[str, Any]: """단위 수 × 단위당 정상 호출 — 이미지 문과 글 문을 **따로** 낸다.""" units = expected_units_of(run_id) by = {} for s, n in units.items(): img_n, _ = image_calls_per_unit_of(s) by[s] = {"units": n, "images": n * img_n, "text": n * normal_calls_per_unit_of(s) + extra_calls_per_run_of(s)} return {"by_step": by, "images": int(sum(v["images"] for v in by.values())), "text": int(sum(v["text"] for v in by.values()))} def assert_outlook_pair_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """fixture `OUTLOOK_PAIR_CAP` 이 production 산출(`outlook_phase3` CP 의 아웃룩 수)을 덮나. ★아웃룩 하나 = 쌍 하나(각 아웃룩은 한 인물의 것). null 아웃룩은 production 이 건너뛰므로 여기서는 **위쪽 상한**으로 전부 센다 — 글자로 안 가른다.""" steps = [s for s in metered if METERING_UNITS.get(s) == UNIT_OUTLOOK_PAIR and s not in done] if not steps: return {"checked": False, "why": "아웃룩 쌍 단위 스텝이 살 자리에 없다"} root = ci.root_dir(run_id) cps = sorted((root / "projects").glob("*/checkpoints/episodes/*/outlook_phase3/manifest.json")) if not cps: return {"checked": False, "why": "outlook_phase3 CP 가 아직 없다 — 새 run 은 선언값을 쓴다", "outlook_pair_cap": dims.get("outlook_pair_cap"), "steps": steps} d = json.loads(cps[-1].read_text(encoding="utf-8")) or {} n = len(((d.get("data") or {}).get("outlooks")) or []) cap = int(dims.get("outlook_pair_cap") or 0) if n > cap: raise ScopeMismatch( f"아웃룩 쌍 단위 스텝 {steps} 가 살 자리인데 실제 아웃룩 {n} 이 원고 선언 " f"OUTLOOK_PAIR_CAP {cap} 을 넘는다 — 상수를 원고에 맞춰라") return {"checked": True, "outlooks": n, "outlook_pair_cap": cap, "steps": steps} def assert_outdoor_group_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """fixture `OUTDOOR_GROUP_CAP` 이 production 산출(`background_classify` CP 의 **실외 그룹** 수)을 덮나 — 술어는 production 의 `outdoor_groups_of` 한 곳. ★그룹 수는 계측·상한의 분모이고 보충 행의 분모는 unique location 이라 둘을 **같이** 낸다(Codex 2026-09-03). CP 가 아직 없으면(첫 판) 선언값을 쓰고, 스텝 직전 문(`producer_cap_gate_before`)이 CP 가 생긴 뒤 **provider 앞에서** 다시 센다.""" from app.core.steps.outdoor_place_spec_step import outdoor_groups_of, outdoor_loc_ids_of steps = [s for s in metered if METERING_UNITS.get(s) == UNIT_OUTDOOR_GROUP and s not in done] if not steps: return {"checked": False, "why": "야외 그룹 단위 스텝이 살 자리에 없다"} root = ci.root_dir(run_id) cps = sorted((root / "projects").glob("*/checkpoints/episodes/*/background_classify/manifest.json")) if not cps: return {"checked": False, "why": "background_classify CP 가 아직 없다 — 새 run 은 선언값을 쓴다", "outdoor_group_cap": dims.get("outdoor_group_cap"), "steps": steps} d = json.loads(cps[-1].read_text(encoding="utf-8")) or {} n = len(outdoor_groups_of(d)) locs = outdoor_loc_ids_of(d) cap = int(dims.get("outdoor_group_cap") or 0) if n > cap: raise ScopeMismatch( f"야외 그룹 단위 스텝 {steps} 가 살 자리인데 실제 실외 그룹 {n} (unique location {len(locs)}) 이 " f"원고 선언 OUTDOOR_GROUP_CAP {cap} 을 넘는다 — 사지 않고 선다") return {"checked": True, "outdoor_groups": n, "unique_outdoor_locations": len(locs), "outdoor_loc_ids": locs, "outdoor_group_cap": cap, "steps": steps} def _last_cp_data(run_id: str, step: str) -> Optional[Dict[str, Any]]: root = ci.root_dir(run_id) cps = sorted((root / "projects").glob(f"*/checkpoints/episodes/*/{step}/manifest.json")) if not cps: return None return (json.loads(cps[-1].read_text(encoding="utf-8")) or {}).get("data") or {} def _assert_unit_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str], *, unit: str, cap_key: str, producer_step: str, count_fn, what: str) -> Dict[str, Any]: """배경 사슬 세 단위의 공통 뼈대 — fixture 상한(`dims[cap_key]`)이 producer CP 를 production 술어로 센 수를 덮나.""" steps = [s for s in metered if METERING_UNITS.get(s) == unit and s not in done] if not steps: return {"checked": False, "why": f"{what} 단위 스텝이 살 자리에 없다"} data = _last_cp_data(run_id, producer_step) if data is None: return {"checked": False, "why": f"{producer_step} CP 가 아직 없다 — 새 run 은 선언값을 쓴다", cap_key: dims.get(cap_key), "steps": steps} n = int(count_fn(data)) cap = int(dims.get(cap_key) or 0) if n > cap: raise ScopeMismatch(f"{what} 단위 스텝 {steps} 가 살 자리인데 실제 {what} {n} 이 원고 선언 {cap_key.upper()} {cap} 을 넘는다 — 사지 않고 선다") return {"checked": True, what: n, cap_key: cap, "steps": steps} def assert_chain_group_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """fixture `CHAIN_GROUP_CAP` 이 `background_classify` CP 의 chain_bg 그룹 수(술어 `chain_groups_of` 한 곳)를 덮나.""" from app.core.steps.background_master_plan_step import chain_groups_of return _assert_unit_cap_covers(run_id, dims, done, metered, unit=UNIT_CHAIN_GROUP, cap_key="chain_bg_group_cap", producer_step="background_classify", count_fn=lambda d: len(chain_groups_of(d.get("building_groups") or [])), what="chain_bg_groups") def assert_floor_plan_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """fixture `FLOOR_PLAN_CAP` 이 `background_master_plan` CP 의 floor plan 수(`floor_plans_of`)를 덮나 — 글·이미지 두 스텝의 단위.""" from app.modules.pipeline.background_master_plan import floor_plans_of return _assert_unit_cap_covers(run_id, dims, done, metered, unit=UNIT_FLOOR_PLAN, cap_key="floor_plan_cap", producer_step="background_master_plan", count_fn=lambda d: len(floor_plans_of(d.get("plans") or {})), what="floor_plans") def assert_background_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """fixture `BACKGROUND_CAP` 이 `background_master_plan` CP 의 background 수(`backgrounds_of`)를 덮나.""" from app.modules.pipeline.background_master_plan import backgrounds_of return _assert_unit_cap_covers(run_id, dims, done, metered, unit=UNIT_BACKGROUND, cap_key="background_cap", producer_step="background_master_plan", count_fn=lambda d: len(backgrounds_of(d.get("plans") or {})), what="backgrounds") def assert_central_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """계획이 잡은 중앙 대상 수(`dims.central_target_count`)가 production 이 **실제로 세운 의무 수** (`grounding_screen` CP `counts.obligation`)를 덮나 — provider 앞에서 센다. ★★★실측 4398a55dc0bb 2판(Codex 긴급 BLOCK 2026-09-03 05:08): dry 때 screen CP 가 없어 선언 하한 5 로 상한 25 를 열었는데, 같은 live 안에서 screen 이 의무 **20** 을 세웠다 — 이대로면 다섯 대상 뒤 step cap 에서 선다(2026-09-02 밤 attempt 21c47e1b 와 같은 부류). 넘으면 사지 않고 서고, dry 를 다시 내어 상한을 재산정한다.""" steps = [s for s in metered if METERING_UNITS.get(s) == UNIT_CENTRAL and s not in done] if not steps: return {"checked": False, "why": "중앙 단위 스텝이 살 자리에 없다"} actual = obligation_count_of(run_id) planned = dims.get("central_target_count") if actual is None: return {"checked": False, "why": "grounding_screen CP 가 아직 없다 — 새 run 은 선언값을 쓴다", "central_target_count": planned, "steps": steps} if planned is None: raise ScopeMismatch(f"중앙 스텝 {steps} 가 살 자리인데 계획 근거(central_target_count)가 없다 — 사지 않고 선다") if int(actual) > int(planned): # ★★계획이 작았다(새 run 은 screen CP 가 없어 선언 하한으로 잡는다). 서지 않고 **실제 의무 수로 step cap 을 재산정**해 # 이어간다(Codex 2026-09-03 06:05 「비용 때문에 다시 멈추지 말 것」). 승인 정지선(counted 문)은 그대로라 돈은 그 안이다. # 원고(fixture)가 아니라 production 산출이 분모이므로 사람 선언을 넘는 것이 아니다 — 그래서 여기만 재산정이고 # fixture 상한(outdoor_group · outlook 등)을 넘는 것은 여전히 선다. fixture = str(dims.get("fixture") or "") cap = central_logical_cap(fixture, obligations=int(actual)) if fixture else None if cap is None: raise ScopeMismatch( f"중앙 스텝 {steps} 가 살 자리인데 production 의무 {actual} 이 계획 대상 수 {planned} 를 넘고 원고 이름이 없어 " "재산정을 못 한다 — 사지 않고 선다") return {"checked": True, "obligations": int(actual), "central_target_count": int(planned), "recomputed_logical_cap": int(cap["logical_cap"]), "recomputed_target_count": int(cap["target_count"]), "why": f"계획 {planned} < 실제 의무 {actual} — step cap 을 {cap['logical_cap']} 으로 재산정(정지선은 그대로)", "steps": steps} return {"checked": True, "obligations": int(actual), "central_target_count": int(planned), "steps": steps} def assert_state_variant_cap_covers(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """fixture `STATE_VARIANT_CAP` 이 production 산출(`shot_staging` CP 의 (인물, 상태))을 덮나. ★상태 판정은 production 의 `is_immobilized_state` 한 곳 — 글자로 안 가른다. 이름→엔티티 매핑은 production 이 더 줄이므로 여기 수는 **위쪽 상한**이다.""" from app.core.subject_state import is_immobilized_state steps = [s for s in metered if METERING_UNITS.get(s) == UNIT_STATE_VARIANT and s not in done] if not steps: return {"checked": False, "why": "상태 변형 단위 스텝이 살 자리에 없다"} root = ci.root_dir(run_id) cps = sorted((root / "projects").glob("*/checkpoints/episodes/*/shot_staging/manifest.json")) if not cps: return {"checked": False, "why": "shot_staging CP 가 아직 없다 — 새 run 은 선언값을 쓴다", "state_variant_cap": dims.get("state_variant_cap"), "steps": steps} d = json.loads(cps[-1].read_text(encoding="utf-8")) or {} pairs = set() for shot in ((d.get("data") or d).get("shots") or []): for ca in (shot.get("character_angles") or []): state = ca.get("subject_state") if state is not None and is_immobilized_state(str(state)): pairs.add((str(ca.get("character") or ""), str(state))) cap = int(dims.get("state_variant_cap") or 0) if len(pairs) > cap: raise ScopeMismatch( f"상태 변형 단위 스텝 {steps} 가 살 자리인데 실제 (인물, 상태) {len(pairs)} 가 " f"원고 선언 STATE_VARIANT_CAP {cap} 을 넘는다 — 상수를 원고에 맞춰라") return {"checked": True, "state_variants": len(pairs), "state_variant_cap": cap, "steps": steps} # ★★★producer 산출을 읽는 상한 문 — 단위별 하나. `run_pipeline(before_step=)` 이 **그 단위의 첫 스텝을 # 부르기 직전**(provider 앞)에 부른다. 첫 판엔 CP 가 없어 시작 전 preflight 는 선언값만 보므로, 이 문이 # 없으면 셋째 그룹은 앞 둘을 산 뒤 provider 경계에서 선다(Codex BLOCK 2026-09-03). PRODUCER_CAP_GATES = { UNIT_OUTDOOR_GROUP: "assert_outdoor_group_cap_covers", UNIT_OUTLOOK_PAIR: "assert_outlook_pair_cap_covers", UNIT_STATE_VARIANT: "assert_state_variant_cap_covers", UNIT_ENTITY: "assert_entity_cap_covers_queue", UNIT_CENTRAL: "assert_central_cap_covers", UNIT_CHAIN_GROUP: "assert_chain_group_cap_covers", UNIT_FLOOR_PLAN: "assert_floor_plan_cap_covers", UNIT_BACKGROUND: "assert_background_cap_covers", } def producer_cap_gate_before(step: str, *, run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Optional[Dict[str, Any]]: """스텝 `step` 을 부르기 직전 — 그 스텝의 단위에 producer 상한 문이 있으면 **지금 있는 CP** 로 센다. 넘으면 `ScopeMismatch` (사지 않고 선다). 문이 없는 단위·이미 끝난 스텝은 None.""" unit = METERING_UNITS.get(step) name = PRODUCER_CAP_GATES.get(unit or "") if not name or step not in set(metered): return None # ★부르는 쪽(run_pipeline)이 **살 스텝**에만 부른다 — 끝났어도 지문 어긋남으로 force 되는 스텝은 산다. # 그래서 `done` 에서 이 스텝을 빼고 센다(실측 4398a55dc0bb 재개: 야외 스텝이 completed 인 채 force 였다). fn = globals()[name] got = fn(run_id, dims, set(done) - {step}, metered) return {"unit": unit, "gate": name, **dict(got or {})} def assert_entity_cap_covers_queue(run_id: str, dims: Dict[str, Any], done: set, metered: Sequence[str]) -> Dict[str, Any]: """원고가 선언한 ENTITY_CAP 이 production 의 실제 entity_queue 를 **덮는가**. Codex 재리뷰 (2026-09-02): fixture 상수로 적는다면 production queue 와 대조하는 문이 있어야 한다 — 원고를 바꾸고 상수를 안 바꾸면 여기서 선다. 엔티티 단위 스텝이 살 자리에 있고 `entity_detail` CP 가 있을 때만 잰다. """ entity_steps = [s for s in metered if METERING_UNITS.get(s) == UNIT_ENTITY and s not in done] if not entity_steps: return {"checked": False, "why": "엔티티 단위 스텝이 살 자리에 없다"} root = ci.root_dir(run_id) qs = sorted((root / "projects").glob("*/checkpoints/episodes/*/entity_detail/manifest.json")) if not qs: return {"checked": False, "why": "entity_detail CP 가 아직 없다 — 새 run 은 선언값을 쓴다", "entity_cap": dims.get("entity_cap"), "steps": entity_steps} d = json.loads(qs[-1].read_text(encoding="utf-8")) or {} n = len(((d.get("data") or {}).get("entity_queue")) or []) cap = int(dims.get("entity_cap") or 0) if n > cap: raise ScopeMismatch( f"엔티티 단위 스텝 {entity_steps} 가 살 자리인데 실제 queue {n} 이 원고 선언 " f"ENTITY_CAP {cap} 을 넘는다 — 상수를 원고에 맞춰라") return {"checked": True, "queue": n, "entity_cap": cap, "steps": entity_steps} def unverified_units_in(steps: Sequence[str]) -> List[str]: """closure 안의 계측 스텝 중 단위가 `unverified` 인 것.""" return [s for s in steps if METERING_UNITS.get(s) == UNIT_UNVERIFIED] def build_plan(sc=None, *, run_id: Optional[str] = None) -> Dict[str, Any]: """실행계획과 상한. ★손으로 적은 수 위에 승인이 선다. ★세 축은 **시나리오 하나**에서 온다 — 기본값으로 따로 안 떨어진다. ★`run_id` 를 주면 중앙 조사 상한을 그 run 의 **실제 의무 수**로 잰다(선언은 하한). """ from app.core.step_manifest import STEP_MANIFEST as M sc = _scenario(sc) # ★기본 모드면 **모듈 선언**을 쓴다 — 그래야 시험이 그것을 갈아 끼워 # 「선언이 production 과 다르면 선다」를 잴 수 있다. cfg = fixture_config(sc["mode"], fixture=sc["fixture"]) dims = fixture_dimensions(sc["fixture"]) plan_basis: Dict[str, Any] = {} # 계획 근거(run 상태에서 온 값) — 원고 치수와 섞지 않는다 plan = ct.execution_plan(config=cfg, target=sc["target"]) plan["contract"] = dict(LOCKED_CONTRACT) plan["fixture_dimensions"] = dims plan["target"] = sc["target"] metered = plan["applied_metered"] # ★단위는 `logical_cap_of` **한 곳**이 정한다 — 여기서 다시 안 적는다 # ★정상 예상(단위당 1)과 hard cap(단위 × 허용 호출)을 **갈라** 낸다 normal = {s: logical_cap_of(s, dims) * normal_calls_per_unit_of(s) + extra_calls_per_run_of(s) for s in metered} caps = {s: hard_cap_of(s, dims) + extra_calls_per_run_of(s) for s in metered} # ★중앙 조사만 **전용 상한** — 대상 수에서 나온다(일반 1 이 아니다). # ★그 스텝이 **도는 판에서만** 구한다 — legacy 는 아예 안 돌린다. central = None if CENTRAL_STEP in caps: central = central_logical_cap(sc["fixture"], obligations=obligation_count_of(run_id)) caps[CENTRAL_STEP] = int(central["logical_cap"]) # ★계획이 무엇을 근거로 중앙 상한을 잡았는지 — 스텝 직전 문(`assert_central_cap_covers`)이 실제 의무 수와 견준다. # ★dims(원고 치수)에 넣지 않는다 — `assert_scope` 가 판마다 dims 를 대조하므로 run 상태에서 온 값이 섞이면 # 같은 run 의 재개가 「치수가 다르다」로 선다(실측 05:14). 따로 `plan_basis` 로 낸다. plan_basis["central_target_count"] = int(central["target_count"]) plan_basis["central_obligation_count"] = central.get("obligation_count") plan_basis["fixture"] = str(sc["fixture"]) # ★문이 실제 의무 수로 재산정할 때 쓴다 normal[CENTRAL_STEP] = int(central["logical_cap"]) rows = [r for r in ct.plan_rows(logical_caps=caps, contract=LOCKED_CONTRACT, target=sc["target"]) if r["step"] in set(metered)] tot = ct.totals(rows) # ★승인선과 견주는 것은 **정상 예상**이다. hard cap 합은 따로 적는다 — # 그 합은 재시도가 다 터졌을 때의 스텝 문 합이고, run 전체 정지선이 먼저 선다. tot["logical_cap_total"] = int(sum(normal.values())) tot["hard_cap_total"] = int(sum(caps.values())) for r in rows: r["normal_expected"] = int(normal.get(r["step"], 0)) r["calls_per_unit"] = calls_per_unit_of(r["step"]) return {"plan": plan, "caps": caps, "normal": normal, "totals": tot, "rows": rows, "dimensions": dims, "plan_basis": plan_basis, # ★단위를 안 읽은 스텝 — 살 자리에 있으면 live 가 선다 "unverified_units": unverified_units_in(metered), "scenario": dict(sc), "fixture_config": cfg, "central_cap": central, # ★이미지 문의 표 — 글 표와 **따로** (같은 dims · 같은 단위) "images": image_plan(dims, metered), "bootstrap": ct.bootstrap_rows(contract=LOCKED_CONTRACT)} def run_outputs(run_id: str) -> List[Path]: """이 run 이 **실제로 돈 판**의 산출들. `[첫 판, 재개1, 재개2, …]`. ★재개는 제 파일을 따로 쓴다(`canary_resume__.json`) — 첫 판을 안 덮으려고. 그래서 「지금 어느 코드에 서 있나」는 **가장 뒤 파일**이 안다. 순서는 mtime 이 아니라 **번호**로 센다. """ root = ci.root_dir(run_id) first = root / "canary_run.json" if not first.is_file(): raise ScopeMismatch(f"run {run_id} 의 산출이 없다 — 앞 tip 을 모른다") def _n(p: Path) -> int: m = re.search(r"_(\d+)\.json$", p.name) return int(m.group(1)) if m else 0 return [first] + sorted(root.glob("canary_resume_*.json"), key=_n) def recorded_tip(run_id: str) -> str: """이 run 이 **마지막으로 실제로 돈** 코드 신원. ★손으로 안 넘긴다. ★★★실측 (2026-09-02): 내가 `from_tip` 을 인자로 받게 해 놓고 짧은 해시만 알고 **뒤를 지어냈다** — 장부에 없는 SHA 가 적혔다. 값이 이미 산출에 있는데 사람이 다시 타이핑하게 두면 그런 일이 난다. ★★★두 번째 실측 (2026-09-02): 앞 판은 **첫 판 산출만** 읽었다. 그래서 `A→B` 를 적고 B 에서 재개한 뒤 결함을 하나 더 고쳐 C 를 만들면 「앞 tip 이 A 인데 B 로 알고 있다」로 서서, 얼마 썼는지를 들고 있는 run 을 **버려야** 했다. 이제 마지막으로 **돈** 판을 읽는다 — 전이 줄을 따라가면 안 된다(그러면 머리가 곧 목적지가 되어 전이 조건 검사가 통째로 건너뛰어진다. 실측으로 겪었다). """ import json as _json p = run_outputs(run_id)[-1] got = str(((_json.loads(p.read_text(encoding="utf-8")) or {} ).get("code") or {}).get("tip") or "") if not got: raise ScopeMismatch(f"run {run_id} 산출 {p.name} 에 코드 tip 이 없다") return got def code_lineage(run_id: str) -> List[str]: """이 run 이 **실제로 돌아 온** 코드들. `[처음, …, 지금]` (중복 접음).""" import json as _json out: List[str] = [] for p in run_outputs(run_id): tip = str(((_json.loads(p.read_text(encoding="utf-8")) or {} ).get("code") or {}).get("tip") or "") if tip and (not out or out[-1] != tip): out.append(tip) return out def run_identity(run_id: str) -> Optional[Dict[str, str]]: """이 run 의 **신원** — 부트스트랩 기록(`canary_run.json`)의 mode·fixture. ★옛 기록에 scenario 가 없으면 None(대조 못 함). target·lane 축은 판마다 바뀔 수 있어 신원이 아니다.""" p = ci.root_dir(run_id) / "canary_run.json" if not p.is_file(): return None sc = (json.loads(p.read_text(encoding="utf-8")) or {}).get("scenario") or {} if not sc.get("mode") or not sc.get("fixture"): return None return {"mode": str(sc["mode"]), "fixture": str(sc["fixture"])} def assert_scenario_matches_run(run_id: str, sc: Dict[str, Any]) -> Optional[Dict[str, str]]: """지금 시나리오의 mode·fixture 가 **이 run 의 신원**과 같은가. ★★실측 (2026-09-02): 전이를 적으며 `--fixture` 를 빠뜨려 CLI 기본값 `canary_one_scene` 이 period_episode run 의 장부에 적혔다(append-only 라 supersedes 로 정정). 값을 사람이 다시 타이핑하게 두면 기본값이 스며든다 — run 이 이미 아는 신원과 다르면 **선다**. """ ident = run_identity(run_id) if ident is None: return None for k in ("mode", "fixture"): if str(sc.get(k)) != ident[k]: raise ScopeMismatch( f"이 run 은 {ident['mode']}·{ident['fixture']} 인데 지금 시나리오는 " f"{sc.get('mode')}·{sc.get('fixture')} 다 — 다른 run 의 신원으로 안 잇는다" f"(CLI 기본값이 스며들었나 보라)") return ident def record_code_transition(run_id: str, *, why: str, sc: Dict[str, Any], from_tip: Optional[str] = None, hash_adoptions: Sequence[str] = ()) -> Dict[str, Any]: """같은 run 을 **다른 코드로** 이어 갈 때 그 전이를 장부에 남긴다. ★★★일반 범위 문을 느슨하게 하지 않는다 — 이 run 에 **한해** 전이를 `append-only` 로 승인·보존한다 (Codex BLOCK 2026-09-02). 앞 판의 top-level 증거는 한 글자도 안 덮는다. Raises: ScopeMismatch: 지금 트리가 더럽다 · `from_tip` 이 아니다 · 열린 attempt 가 있다 · `grounding_chunk` 체크포인트가 이미 있다. """ from tools.grounding_audit import canary_pipeline as cp # ★hash_adoptions — 이 전이에 **결속된** adoption 사건들(`canary_hash_adoption.adopt` 가 적은 event_id). 그 사건은 # 지금 tip(to_tip)의 코드로 계산한 새 hash 를 옛 CP 에 **data 불변**으로 입힌 것이라, 사건의 at_tip 이 to_tip 과 같아야 한다. _adopt = sorted({str(x).strip() for x in (hash_adoptions or ()) if str(x).strip()}) # ★앞 tip 은 **run 산출에서** 읽는다. 인자로 준 것은 대조용이다. # ★★마지막으로 **돈** 판의 tip 이다 — 이미 한 번 이어서 돌았으면 # 그 코드에서 다음으로 잇는다. real = recorded_tip(run_id) assert_scenario_matches_run(run_id, sc) # ★run 의 신원(mode·fixture)과 대조 if from_tip is not None and not real.startswith(str(from_tip)): raise ScopeMismatch( f"앞 tip 이 {real} 인데 {from_tip!r} 로 알고 있다 — " f"모르는 채 안 잇는다") from_tip = real now = git_tip() if not now["clean"]: raise ScopeMismatch( f"트리가 더럽다 {now['dirty_files'][:5]} — 무슨 코드로 이어 가는지 " f"못 적는다") if now["tip"] == from_tip: raise ScopeMismatch( f"tip 이 그대로다({from_tip[:8]}) — 전이가 아니다") root = ci.root_dir(run_id) if _adopt: _events = {str(r.get("event_id")): r for r in cp.read_attempts(root) if r.get("kind") == cp.EVENT_HASH_ADOPTION} for eid in _adopt: ev = _events.get(eid) if ev is None: raise ScopeMismatch(f"hash adoption 사건 {eid} 가 이 run 의 장부에 없다") if str(ev.get("at_tip")) != now["tip"]: raise ScopeMismatch( f"adoption {eid} 는 {str(ev.get('at_tip'))[:8]} 코드로 계산했는데 지금은 {now['tip'][:8]} 다 — 다른 코드의 hash 를 잇지 않는다") live = cp.open_attempts(root) if live: raise ScopeMismatch( f"열린 attempt 가 {len(live)}개다 — 얼마 썼는지 모르는 채 못 잇는다") rec = cp.append_event(root, { "kind": cp.EVENT_CODE_TRANSITION, "from_tip": from_tip, "to_tip": now["tip"], "why": why, "hash_adoptions": _adopt, "scenario": {k: sc[k] for k in _SCENARIO_AXES}, "approved": approved_for(sc["mode"]), "approved_images": approved_images_for(sc), # ★이번 attempt 의 실제 문 — 표의 10/40 을 재승인한 것으로 안 읽히게 "outbound_doors_this_attempt": { k: v for k, v in outbound_doors_on_resume( run_id, approved_for(sc["mode"])).items() if k in ("search", "download", "★doors_note")}, "dimensions": fixture_dimensions(sc["fixture"]), "locks": dict(LOCKED_CONTRACT), "cumulative_used": cp.cumulative_used(root), "diff_stat": _diff_stat(from_tip, now["tip"]), "★means": ("같은 run 을 다른 코드로 잇는다. 앞 판 증거는 그대로 두고 " "**덧붙인다** — 무엇이 언제 바뀌었는지가 남아야 한다"), }) return rec def _diff_stat(a: str, b: str) -> Dict[str, Any]: """두 tip 사이에 **무엇이** 달라졌나. ★요약이 아니라 파일 목록이다.""" out = subprocess.run(["git", "diff", "--stat", f"{a}..{b}"], cwd=str(BACKEND.parent), capture_output=True, text=True) names = subprocess.run(["git", "diff", "--name-only", f"{a}..{b}"], cwd=str(BACKEND.parent), capture_output=True, text=True) return {"files": [x for x in names.stdout.splitlines() if x], "stat": out.stdout.strip().splitlines()[-1:]} def central_calls_per_target_per_round() -> int: """중앙 조사가 **한 대상 한 라운드**에 부르는 counted 글/VLM 호출 수. ★★★**production 이 실제로 배선한 것만** 센다 (실측 2026-09-02). 앞 판은 `acquire_one` 소스에 `write_brief` 와 `judge` 가 보인다고 **2** 로 적었다. 그런데 production(`grounding_central_acquisition`)은 `write_brief` 를 **안 넘긴다** — `acquire_one` 안에서 `if write_brief is not None` 로 건너뛴다. 즉 그 갈래는 안 돈다. 「함수에 있다」와 「production 이 부른다」는 다르다. judge 라운드마다 1 — 심판 안 `call_structured` 가 정확히 1개 write_brief production 이 **안 넘기면 0** `search`·`download` 는 이미지 검색 갈래라 글 예산과 **따로** 센다. """ import ast import inspect import textwrap from app.modules.pipeline import grounding_central_acquisition as ca tree = ast.parse(textwrap.dedent(inspect.getsource(ca.run))) wired = set() for n in ast.walk(tree): if (isinstance(n, ast.Call) and ast.unparse(n.func).endswith("acquire_one")): wired |= {k.arg for k in n.keywords} if not wired: raise ScopeMismatch( "중앙이 `acquire_one` 을 부르는 자리를 못 찾았다 — 상한을 " "지어내지 않는다") return 1 + (1 if "write_brief" in wired else 0) #: ★뒤로 호환 — 부르는 쪽·시험이 이 이름을 쓴다. 값은 위 함수가 낸다. CENTRAL_CALLS_PER_TARGET_PER_ROUND = None def _owner_requirement(fixture: str) -> Dict[str, Any]: """중앙에 넘길 요구. ★선언이 없는 원고면 **아무것도 안 건다**. ★legacy 회귀 fixture 는 갈래를 선언하지 않는다 — 거기에 없는 요구를 지어 넣지 않는다. """ out: Dict[str, Any] = {} try: out["required_owner_types"] = required_owner_types(fixture) except ScopeMismatch: pass # ★★원고가 선언한 시대·지역을 **구조화 칸**으로 넘긴다 (Codex BLOCK 2026-09-02) — # `visual_world_rules` 가 그 글자를 그대로 쓴다. 빈 선언(현대)은 안 넘긴다. from app.modules.pipeline.grounding_coordinates import ( DECLARED_ERA_KEY, DECLARED_REGION_KEY) fx = cbs.load_fixture(fixture) for key, attr in ((DECLARED_ERA_KEY, "ERA"), (DECLARED_REGION_KEY, "REGION")): v = str(getattr(fx, attr, "") or "").strip() if v: out[key] = v return out def required_owner_types(fixture: str) -> List[str]: """이 원고가 **지나야 한다고 선언한** 갈래들. ★이름이 아니라 owner 다.""" fx = cbs.load_fixture(fixture) got = {str(t.get("owner")) for t in (getattr(fx, "EXPECTED_TARGETS", None) or ()) if str(t.get("key", "")).startswith("owner_") and t.get("owner")} if not got: raise ScopeMismatch(f"원고 {fixture!r} 가 갈래를 선언하지 않는다") return sorted(got) def obligation_count_of(run_id: Optional[str]) -> Optional[int]: """그 run 이 **실제로 세운 조사 의무 수** — `grounding_screen` CP 의 `counts.obligation`. CP 가 없으면(첫 판 · dry) `None` — 그때는 원고 선언이 하한이다.""" if not run_id: return None root = ci.root_dir(run_id) for p in root.glob("projects/*/checkpoints/episodes/*/grounding_screen/manifest.json"): try: counts = ((json.loads(p.read_text(encoding="utf-8")) or {}).get("data") or {}).get("counts") or {} except Exception: # noqa: BLE001 return None got = counts.get("obligation") return int(got) if isinstance(got, int) and not isinstance(got, bool) else None return None def central_logical_cap(fixture: str, *, rounds: Optional[int] = None, per_round: Optional[int] = None, obligations: Optional[int] = None) -> Dict[str, Any]: """중앙 조사 스텝의 **전용 논리 상한**. ★일반 non-fanout 1 을 안 쓴다. ★★★`reference_acquisition` 은 fan_out 이 아니라 **대상 수만큼** 돈다. 일반 규칙(비-fanout=1)을 쓰면 `cap=2` 가 되어 첫 대상 언저리에서 끝나고 나머지는 **예산 때문에** `reference_unavailable` 로 접힌다 — 그러면 coverage·terminal 은 초록인데 「다섯 갈래가 중앙 조사 문을 지났다」는 canary ①의 목적을 **못 잰다** (Codex BLOCK 2026-09-02). 대상 수는 **원고가 선언한 것**에서 온다 — 모델 산출이 아니다. """ from app.core.steps.reference_acquisition_step import CENTRAL_ROUNDS fx = cbs.load_fixture(fixture) declared = getattr(fx, "EXPECTED_TARGETS", None) if not declared: raise ScopeMismatch( f"원고 {fixture!r} 가 기대 대상을 선언하지 않는다 — 중앙 조사 " f"상한을 지어내지 않는다") # ★조사 대상은 원고의 **의미 선언**(`AXIS_BASIS[key]["hard"]` — 모델이 틀릴 만한가)으로 # 센다. 앞 판은 열쇠 이름의 접두 `owner_` 로 셌다 — modern_episode 는 열쇠가 # `fuel_station`·`store` 같은 뜻 이름이라 「갈래 대상 선언이 없다」로 섰다 # (실측 2026-09-02). 이름이 아니라 뜻으로 가른다. 선언이 없으면 옛 접두로. basis = getattr(fx, "AXIS_BASIS", None) or {} if basis: owners = [t for t in declared if bool((basis.get(str(t.get("key", ""))) or {}).get("hard"))] else: owners = [t for t in declared if str(t.get("key", "")).startswith("owner_")] if not owners: raise ScopeMismatch(f"원고 {fixture!r} 에 갈래 대상 선언이 없다") r = int(CENTRAL_ROUNDS if rounds is None else rounds) each = int(central_calls_per_target_per_round() if per_round is None else per_round) from app.modules.pipeline import coarse_type_pick as ctp # ★★★원고 선언은 **하한**이다 (실측 2026-09-02 밤, attempt 21c47e1b): 원고는 갈래 다섯을 # 선언했는데 production 은 의무 **21** 을 세웠고, 5 로 잰 상한 40 에 조사가 통째로 죽었다 # (유료 40 쓴 뒤). 모집단은 production 함수가 센 것(`grounding_screen.counts.obligation`) # 으로, 선언보다 작게는 안 잡는다. 재판정(`rejudge_rows`) 도 같은 상한을 쓰므로 # 대상당 1 을 더한다. n_declared = len(owners) n = max(n_declared, int(obligations or 0)) return { "target_count": n, "declared_target_count": n_declared, "obligation_count": obligations, "rounds": r, "per_round": each, "rejudge_each": 1, "one_round_each": n * each, "logical_cap": n * (each * r + 1), # ★★★**세 문이 서로 다른 갈래다** (Codex 2026-09-02). # 글/VLM 예산은 검색 요청도 다운로드도 **못 센다**. # 한 대상 한 라운드가 여는 것 — `acquire_one` 을 읽어 셌다: # `write_brief` 1 + `judge` 1 → counted 글/VLM # `search` 1 → 검색 **요청** # `download` ≤ PER_ROUND_CAP → 받는 **장수** "search_requests_cap": n * r, "downloads_cap": n * r * int(ctp.PER_ROUND_CAP), "candidates_per_round": int(ctp.PER_ROUND_CAP), "★means": ("대상 수(선언과 production 의무 중 큰 쪽) × (라운드 × 라운드당 호출 + 재판정 1) = 글/VLM 논리. " "검색 **요청**은 대상×라운드로 그보다 훨씬 적고, " "**받는 장수**는 그 × 라운드당 후보 상한이다")} def assert_coordinates_in_queries(rows, *, era: str, region: str ) -> Dict[str, Any]: """나간 질의들이 **선언된 좌표를 다 실었나**. ★기계적으로 센다. Args: rows: 중앙 산출의 줄들 — 각 줄의 `acquisition.rounds[*].queries`. ★선언 안 된 좌표는 **안 본다** — 없는 것을 지어내라고 요구하지 않는다. ★뜻을 판단하지 않는다. 선언된 글자가 질의 안에 있는지만 본다. """ want = {k: v for k, v in (("era", era), ("region", region)) if v} total, missing = 0, [] for r in rows: for rd in ((r.get("acquisition") or {}).get("rounds") or ()): for q in (rd.get("queries") or ()): total += 1 gone = [k for k, v in want.items() if str(v) not in str(q)] if gone: missing.append({"subject": r.get("research_subject_id"), "round": rd.get("round_no"), "query": str(q)[:80], "missing": gone}) return {"declared": want, "queries": total, "missing": missing, "ok": not missing, "★means": ("선언된 좌표만 본다 — 없는 좌표는 요구하지 않는다. " "글자만 보고 뜻은 판단하지 않는다")} def attempt_seq(root) -> int: """이 run 의 **몇 번째 재개**인가. ★파일 이름이 안 겹치게.""" return 1 + len(list(root.glob("canary_resume_*.json"))) def valid_transition(run_id: str, *, to_tip: str, from_tip: str) -> Optional[Dict[str, Any]]: """`from_tip` 에서 `to_tip` 으로 잇는 유효한 전이 **한 줄**. 없으면 `None`. ★★`supersedes` 를 **실제로 해석한다** — 뒤 줄이 앞 줄을 대신한다고 적혀 있으면 앞 줄은 무효다. 안 그러면 내가 잘못 적은 줄도 「있다」로 세어진다 (실제로 가짜 `from_tip` 을 적은 적이 있다). """ from tools.grounding_audit import canary_pipeline as cp rows = [r for r in cp.read_attempts(ci.root_dir(run_id)) if r.get("kind") == cp.EVENT_CODE_TRANSITION] dead = {str(r.get("supersedes")) for r in rows if r.get("supersedes")} live = [r for r in rows # ★자기 자신을 대신한다고 적지는 않는다 — 그러면 둘 다 죽는다 if str(r.get("event_id") or r.get("recorded_kst")) not in dead and str(r.get("to_tip")) == str(to_tip) and str(r.get("from_tip")) == str(from_tip)] if len(live) != 1: return None return live[0] def assert_resume_transition(run_id: str, sc: Dict[str, Any]) -> Dict[str, Any]: """**재개 입구의 문.** ★첫 저장·DB·provider 보다 앞에 선다. 새 run 이면 아무것도 안 한다. 이미 있는 run 을 **다른 코드로** 이으려면 유효한 전이가 정확히 하나 있어야 하고, 그 줄이 적은 시나리오·잠금· 승인선·치수가 지금과 같아야 한다. ★★★`record_code_transition` 을 만들어 놓고 **아무도 안 불렀다** (Codex BLOCK 2026-09-02: 호출자 0). 그래서 CLI 로 같은 run_id 를 다른 코드로 재개해도 아무것도 안 막았다. 이제 이 함수가 `run()` 안에서 **반드시** 불린다. Raises: ScopeMismatch: 트리가 더럽다 · 전이가 없거나 둘 이상이다 · 그 줄의 조건이 지금과 다르다 · 열린 attempt 가 있다. """ from tools.grounding_audit import canary_pipeline as cp root = ci.root_dir(run_id) if not (root / "canary_run.json").is_file(): return {"resume": False, "why": "새 run — 이어 갈 것이 없다"} assert_scenario_matches_run(run_id, sc) # ★같은 코드라도 다른 신원이면 선다 now = git_tip() if not now["clean"]: raise ScopeMismatch( f"트리가 더럽다 {now['dirty_files'][:5]} — 무슨 코드로 잇는지 " f"못 적는다") # ★★마지막으로 **돈** 판의 tip. 전이 줄을 따라가면 안 된다 — # 그러면 머리가 곧 목적지라 아래 조건 검사가 통째로 건너뛰어진다. was = recorded_tip(run_id) live = cp.open_attempts(root) if live: raise ScopeMismatch( f"열린 attempt 가 {len(live)}개다 — 얼마 썼는지 모르는 채 못 잇는다") if was == now["tip"]: return {"resume": True, "same_code": True, "tip": now["tip"], "cumulative": cp.cumulative_used(root)} got = valid_transition(run_id, to_tip=now["tip"], from_tip=was) if got is None: raise ScopeMismatch( f"이 run 은 {was[:8]} 로 돌았는데 지금은 {now['tip'][:8]} 다 — " f"**유효한 전이 기록이 정확히 하나** 있어야 잇는다 " f"(`record_code_transition`). 없거나 둘 이상이면 안 잇는다") if str(got.get("from_tip")) != was: raise ScopeMismatch( f"전이가 {str(got.get('from_tip'))[:8]} 에서 왔다는데 이 run 은 " f"{was[:8]} 로 돌았다") # ★옛 전이 줄(이미지 문 전)은 `approved_images` 칸이 없다 — 그 판들은 승인 0 으로 # 돌았으므로 0 으로 읽는다. 지금 범위에 승인이 있으면(40) 그 줄로는 못 잇는다. got = {**got, "approved_images": int(got.get("approved_images") or 0)} for key, want in (("scenario", {k: sc[k] for k in _SCENARIO_AXES}), ("approved", approved_for(sc["mode"])), ("approved_images", approved_images_for(sc)), ("dimensions", fixture_dimensions(sc["fixture"])), ("locks", dict(LOCKED_CONTRACT))): if got.get(key) != want: raise ScopeMismatch( f"전이 기록의 {key} 가 지금과 다르다 — 적힌 것 {got.get(key)} · " f"지금 {want}") return {"resume": True, "same_code": False, "from_tip": was, "to_tip": now["tip"], "transition": got.get("recorded_kst"), "hash_adoptions": list(got.get("hash_adoptions") or []), "lineage": code_lineage(run_id), "cumulative": cp.cumulative_used(root)} def steps_folding_setting(name: str) -> List[str]: """`_config_hash`/`_config_hash_base` 본문이 그 설정 이름을 읽는 스텝들. ★레지스트리(`STEP_CLASSES`: step_id → 클래스)로 **실제로 도는 클래스**의 소스를 본다 — 파일 이름이나 클래스 이름으로 짐작하지 않는다. ★★★**자식 프로세스에서** 본다 (실측 2026-09-02 격리 near-miss): 이 프로세스에서 `app.core.steps` 를 올리면 `app.core.database` 가 딸려 와 `SessionLocal` 이 **그 순간의 URL** 로 굳는다. 격리 문 뒤에 그런 import 가 있으면 원본 DB 에 붙은 세션을 들고 pipeline 으로 간다. ★정적으로 본다 — 함수 본문에 그 이름이 있는가. 놓치는 것은 runtime 이 두 번째 문으로 잡는다: 해시가 바뀐 스텝은 step_runner 가 RERUN 하려 하고, 앞 판에 끝난 스텝은 cap 0 이라 provider 앞에서 선다. """ code = ( "import inspect, json, sys\n" "from app.core.steps import STEP_CLASSES\n" "name = sys.argv[1]; out = []\n" "for sid, cls in STEP_CLASSES.items():\n" " for fname in ('_config_hash', '_config_hash_base'):\n" " fn = getattr(cls, fname, None)\n" " if fn is None: continue\n" " try: src = inspect.getsource(fn)\n" " except (OSError, TypeError): continue\n" " if name in src and sid not in out: out.append(sid)\n" "print(json.dumps(sorted(out)))\n") got = subprocess.run([sys.executable, "-c", code, str(name)], cwd=str(BACKEND), capture_output=True, text=True, env=dict(os.environ)) if got.returncode != 0: raise ScopeMismatch( f"설정을 접는 스텝을 못 셌다 — 자식 프로세스 실패: " f"{(got.stderr or '')[-400:]}") return list(json.loads(got.stdout.strip().splitlines()[-1])) def partial_steps_of(run_id: str, caps: Dict[str, int]) -> Dict[str, Dict[str, Any]]: """이 run 에서 `partial` 로 끝난 스텝의 **남은 몫**. ★dry 가 이것을 보인다. Codex 재리뷰 (2026-09-02): partial 은 cap 0 되쓰기가 아니다 — 그 스텝의 cap 은 계획표 값 그대로 열리고, 기존 완료분은 다시 안 사며, 예상 신규 = applicable − completed 다. 세 수를 **따로** 적어야 「entity_t2i cap 10 · 기존 12 재구매 0 · 예상 신규 3」처럼 읽힌다. """ root = ci.root_dir(run_id) out: Dict[str, Dict[str, Any]] = {} for m in sorted((root / "projects").glob( "*/checkpoints/episodes/*/*/manifest.json")): try: d = json.loads(m.read_text(encoding="utf-8")) or {} except Exception: continue if str(d.get("status") or "") != "partial": continue s = m.parent.name done = int(d.get("completed_count") or 0) appl = int(d.get("applicable_count") or 0) out[s] = {"cap": int(caps.get(s, 0)), "completed": done, "applicable": appl, "failed": int(d.get("failed_count") or 0), "expected_new": max(appl - done, 0), "★means": "기존 완료분은 다시 안 산다 · 예상 신규는 남은 대상 수"} return out def completed_steps_of(run_id: str, *, statuses: tuple = ("completed", "partial")) -> List[str]: """이 run 에 **끝난 CP 가 있는** 스텝들(기본 completed·partial). ★파일에서 읽는다.""" root = ci.root_dir(run_id) out: List[str] = [] for m in sorted((root / "projects").glob( "*/checkpoints/episodes/*/*/manifest.json")): try: st = str((json.loads(m.read_text(encoding="utf-8")) or {} ).get("status") or "") except Exception: st = "?" if st in statuses: out.append(m.parent.name) return out def drift_reentry_of(run_id: str, candidates: Sequence[str], sc: Dict[str, Any]) -> Dict[str, List[str]]: """끝난(completed) 스텝 중 지금 코드의 지문과 어긋난 것들 — runner 의 그 함수로 묻는다. forced: force 로 다시 산다 — runner 가 하류 CP 를 **지운다**(`forced_would_invalidate_of` 가 그 목록) ★「자기만 다시(하류 보존)」갈래는 없다(Codex BLOCK 2026-09-03). hash 조리법만 바뀐 스텝은 live 전에 `canary_hash_adoption.adopt` 로 어긋남 자체를 없앤다 — 그러면 여기 안 잡히고 cap 0 으로 되쓴다. ★실측 4398a55dc0bb 재개(2026-09-03): 야외 스텝이 completed 인 채 force 였는데 dry 의 남은 표·상한 문·단위 확인이 모두 그것을 done 으로 쳐서 **살 스텝 하나가 계획에서 빠졌다**. 재는 도구가 거짓말한 자리.""" from tools.grounding_audit import canary_pipeline as cp root = ci.root_dir(run_id) eps = sorted((root / "projects").glob("*/checkpoints/episodes/*")) out: Dict[str, List[str]] = {"forced": []} if not eps or not candidates: return out pid, epi = eps[-1].parents[2].name, eps[-1].name cfg = _owner_requirement(sc["fixture"]) covered = cp.code_transition_covers_head(root) for s in sorted(set(candidates)): durable = cp.durable_status(pid, epi, s) if durable != "completed": continue drift = cp.contract_drift_of(s, pid, epi, cfg) if not drift: continue if cp.plan_reentry(durable, drift, covered, False)["mode"] == "force": out["forced"].append(s) return out def forced_for_drift_of(run_id: str, candidates: Sequence[str], sc: Dict[str, Any]) -> List[str]: """`drift_reentry_of(...)["forced"]` (옛 이름).""" return drift_reentry_of(run_id, candidates, sc)["forced"] def forced_would_invalidate_of(run_id: str, forced: Sequence[str], downstream_of: Optional[Callable[[str], Sequence[str]]] = None) -> Dict[str, List[str]]: """force 재실행이 **지울 하류** — runner 의 `invalidate_downstream` 과 같은 술어(`get_all_downstream_recursive`)로, 이 run 에 CP 가 있는 것만. ★dry 의 눈먼 자리(실측 2026-09-03 f7cc45c576c0 dry_28): 「살 자리」표는 지금 있는 CP 로 done 을 세어 force 의 연쇄를 안 보여 줬다 — entity_detail 하나가 force 면 하류 59 중 이 run 의 완료 CP 가 전부 지워진다.""" if not forced: return {} if downstream_of is None: from app.core.step_manifest import get_all_downstream_recursive as downstream_of # noqa: N816 done = set(completed_steps_of(run_id, statuses=("completed", "partial", "not_applicable"))) return {s: sorted(set(downstream_of(s)) & done) for s in forced} def reopened_for_debt_of(run_id: str, candidates: Sequence[str], sc: Dict[str, Any]) -> List[str]: """끝난(completed · 지문 같음) 스텝 중 runner 의 verify_completion 이 **빚이 남았다**는 것들 — 재개가 정상 cap 으로 들어가 빚진 대상만 산다(RERUN_SELF). dry 의 남은 표·상한 문이 이것을 「산다」로 센다.""" from tools.grounding_audit import canary_pipeline as cp root = ci.root_dir(run_id) eps = sorted((root / "projects").glob("*/checkpoints/episodes/*")) if not eps or not candidates: return [] pid, epi = eps[-1].parents[2].name, eps[-1].name cfg = _owner_requirement(sc["fixture"]) out: List[str] = [] for s in sorted(set(candidates)): if cp.durable_status(pid, epi, s) != "completed": continue if cp.contract_drift_of(s, pid, epi, cfg): continue # 지문 어긋남은 forced_for_drift_of 가 센다 if cp.completion_debt_of(s, pid, epi, cfg): out.append(s) return out def previous_axis(run_id: str, axis: str) -> Optional[str]: """앞 판이 그 축을 **어떤 상태로** 돌았나. ★산출에 적힌 것만 읽는다 — 없으면 None (모르면 뒤집힌 것으로 본다). 배경은 옛 산출의 `fixture_config` 도 읽는다.""" p = run_outputs(run_id)[-1] got = json.loads(p.read_text(encoding="utf-8")) or {} sc = got.get("scenario") or {} if sc.get(axis) in ("on", "off"): return str(sc[axis]) if axis == "background": cfg = ((got.get("plan") or {}).get("fixture_config") or {}) if isinstance(cfg.get("background_mode"), bool): return "on" if cfg["background_mode"] else "off" return None def previous_background(run_id: str) -> Optional[str]: """★`previous_axis(run_id, "background")` 의 별칭.""" return previous_axis(run_id, "background") def assert_setting_flip_is_safe(run_id: str, sc: Dict[str, Any] ) -> Dict[str, Any]: """배경 모드를 **뒤집어 같은 run 을 잇는가** — 그러면 끝난 CP 중 그 설정을 `_config_hash` 에 접는 스텝이 **하나도 없어야** 한다 (Codex 조건 2026-09-02). 하나라도 있으면 그 CP 는 새 설정에서 current 가 아니다 — 같은 run 재개가 아니라 별도 run 이어야 한다. ★앞 판 모드를 모르면(None) 뒤집힌 것으로 본다. """ done = completed_steps_of(run_id) out: Dict[str, Any] = {"completed": done, "axes": {}} for axis, spec in _SETTING_AXES.items(): prev = previous_axis(run_id, axis) now = str(sc.get(axis) or axis_actual(axis)) flipped = prev != now folding: List[str] = [] for name in spec["settings"]: folding += [s for s in steps_folding_setting(name) if s not in folding] folding = sorted(folding) stale = sorted(set(folding) & set(done)) if flipped and stale: raise ScopeMismatch( f"{axis} 가 {prev!r} → {now!r} 로 바뀌었는데 끝난 CP 중 " f"{stale} 가 그 설정을 config_hash 에 접는다 — 같은 run 재개가 " "아니다. 별도 run 으로 가라") out["axes"][axis] = {"previous": prev, "now": now, "flipped": flipped, "folding_steps": folding, "stale": stale} # ★옛 호출자(배경만 보던 시험·산출)를 위한 평평한 칸 — 배경 축 그대로 out.update({k: v for k, v in out["axes"]["background"].items()}) return out def outbound_doors_on_resume(run_id: str, approved: Dict[str, Any], reopen: Sequence[str] = (), applied: Optional[Sequence[str]] = None) -> Dict[str, Any]: """이번 attempt 의 검색·받기 문. ★중앙 스텝이 이미 끝났으면 **0** 이다. ★★★Codex BLOCK (2026-09-02): `canary_outbound_scope` 는 attempt 마다 `OutboundBudget` 을 used 0 으로 **새로** 만든다. 「앞 판에서 10/40 을 소진했다」는 이번 문에 안 실린다 — 표의 값을 그대로 넘기면 이번 판이 검색 10·받기 40 을 **다시** 허용한다. 중앙 스텝이 completed 라 cap 0 으로 되쓰이는 것은 정상 경로의 기대일 뿐, 별도 outbound 문이 0 이라는 증거가 아니다. 그래서 여기서 문 자체를 0 으로 연다. ★새 run 이거나 중앙 스텝이 아직 안 끝났으면 표의 값 그대로다. ★★중앙 스텝을 **다시 열었으면** 문도 표의 값이다 — 사람이 열었든(`--steps reference_acquisition`) 코드가 열었든 (지문 어긋남 force · runner 의 빚 RERUN_SELF — 실측 4398a55dc0bb 2026-09-03 05:49: 빚으로 다시 여는 판의 dry 가 「검색 0 · 받기 0」을 냈다). `reopen` 에 그 셋을 합쳐 넘긴다. 실측 2026-09-03 새벽(attempt d75a52a0): 다시 연 조사가 「검색 요청 승인 0」에 막혀 아웃룩 넷· 미완 셋이 전부 빈손으로 끝났다(유료 글 10). 다시 열었다는 것이 재승인이다. """ from app.modules.pipeline.grounding_outbound_consumers import OUTBOUND_CONSUMER_STEPS # ★가벼운 모듈 — 자물쇠 전에 읽는다 done = set(completed_steps_of(run_id, statuses=("completed", "not_applicable"))) reopened = set(reopen or ()) # ★소비자 계약(OUTBOUND_CONSUMER_STEPS) 중 이번 attempt 에 **남았거나 다시 여는** 것 — 하나라도 있으면 표의 값 # ★이 판의 닫힘에 없는 소비자(예: scene_detail 닫힘의 야외 스텝)는 돌지 않는다 — 남은 것으로 세지 않는다 scope = set(applied) if applied is not None else None consumers = [s for s in OUTBOUND_CONSUMER_STEPS if (scope is None or s in scope) and (s not in done or s in reopened)] if consumers: return {**dict(approved), "★outbound_consumers": consumers, "★doors_note": f"검색·받기 소비자 {consumers} 가 남아 있어 표의 값을 연다"} return {**dict(approved), "search": 0, "download": 0, "★outbound_consumers": [], "★doors_note": (f"검색·받기 소비자 {list(OUTBOUND_CONSUMER_STEPS)} 가 전부 completed/not_applicable — 이번 attempt 의 " "검색·받기 문은 0 (표의 값은 재승인이 아니다)")} def assert_scope(built: Dict[str, Any], sc: Dict[str, Any]) -> Dict[str, Any]: """**도는 것이 승인한 것과 같은가.** ★다르면 provider 0 으로 선다. ★세 축 중 **하나만 달라도** 선다 — 한 시나리오로 계획을 짓고 다른 시나리오로 실행하는 일이 없어야 한다 (Codex 조건 ⑥). """ got = built.get("scenario") or {} for axis in _SCENARIO_AXES: if got.get(axis) != sc.get(axis): raise ScopeMismatch( f"계획은 {axis}={got.get(axis)!r} 인데 실행은 " f"{sc.get(axis)!r} 다 — 승인 밖 범위를 살 뻔했다") want = fixture_dimensions(sc["fixture"]) if built["dimensions"] != want: raise ScopeMismatch( f"원고 치수가 다르다 — 계획 {built['dimensions']} · 실제 {want}") if got.get("approved") != approved_for(sc["mode"]): raise ScopeMismatch("승인 정지선이 계획과 다르다") if got.get("approved_images") != approved_images_for(sc): raise ScopeMismatch( f"이미지 승인이 계획과 다르다 — 계획 {got.get('approved_images')!r} · " f"이 범위 {approved_images_for(sc)}") return {"scope": "일치", **{k: sc[k] for k in _SCENARIO_AXES}} def replay_scope(approved: Dict[str, int]) -> Dict[str, Any]: """재판정 판의 **문 셋**. ★검색·받기를 코드로 0 으로 박는다. ★★★운영자가 조심해서 0 을 적는 것이 아니라 **코드가 0 으로 만든다** (Codex 계약 2026-09-02: 「새 이미지 검색 0회·새 다운로드 0장」). 그래야 이 판이 무엇을 못 하는지가 **문에서** 증명된다 — 뭔가 사려 하면 선다. 글/VLM 상한만 남은 것을 그대로 쓴다. """ return {"reopen": (CENTRAL_STEP,), "approved": {**dict(approved), "search": 0, "download": 0}, "★means": ("받아 둔 사진만 다시 판정한다. 검색·받기 문은 0 이라 " "사러 가면 provider 앞에서 선다")} def steps_scope(approved: Dict[str, int], steps: Sequence[str]) -> Dict[str, Any]: """**정한 스텝만** 도는 판의 문 셋. ★검색·받기를 코드로 0 으로 박는다. `replay_scope` 의 일반형이다 — 재판정은 중앙 스텝 하나였고, 이번엔 `shot_selection`·`shot_director` 둘이다(2026-09-02). closure 전부를 돌면 `partial` 로 끝난 스텝이 제 실패를 다시 시도해 문을 세운다(실측). 글/VLM 상한은 **남은 것 그대로** — 이 판이 새 승인선을 여는 것이 아니다. """ want = tuple(str(x) for x in steps if str(x).strip()) if not want: raise ScopeMismatch("도는 스텝을 하나도 안 적었다 — 무엇을 도는지 모른다") # ★★중앙 조사 스텝을 **지목해 다시 열면** 검색·받기 문은 표의 값이다 — 다시 열었다는 것이 # 재승인이다 (실측 2026-09-03 새벽, attempt d75a52a0: 문 0 에 아웃룩 넷·미완 셋이 빈손). 다른 # 스텝만 지목한 판은 여전히 0 — 그 판이 사러 가면 provider 앞에서 선다. doors = ({} if CENTRAL_STEP in set(want) else {"search": 0, "download": 0}) return {"reopen": want, "approved": {**dict(approved), **doors}, "★means": ("적은 스텝만 돈다. 중앙 조사 스텝을 지목했으면 검색·받기 문은 표의 값, " "아니면 0 이라 사러 가면 provider 앞에서 선다")} def run(*, live: bool, run_id: Optional[str] = None, sc: Optional[Dict[str, Any]] = None, replay: bool = False, steps: Sequence[str] = ()) -> Dict[str, Any]: """문 다섯을 **차례로** 지난다. ★하나라도 걸리면 거기서 끝난다. ★★★**격리가 제일 먼저다.** `build_plan()` 은 `app.core.step_manifest` 를 올리는데 그것이 `app.core.database` 를 함께 끌어온다 — 즉 `SessionLocal` 이 **그 시점의 `DATABASE_URL`** 로 만들어진다. 환경을 나중에 바꾸면 이미 **원본에 붙은 세션**을 들고 가게 된다 (2026-08-31 실측). """ # ★★★**무엇보다 먼저** — 여기가 아니면 settings 가 전부 기본값이다 ct.assert_backend_cwd() sc = _scenario(sc) approved = sc["approved"] # ★한 번 읽고 **그 값만** 쓴다 #: 재판정 판이면 **검색·받기 문이 0** 이고 중앙 스텝을 다시 연다. if replay and steps: raise ScopeMismatch("`replay` 와 `steps` 를 같이 줄 수 없다 — 한 판은 한 뜻") _replay = (replay_scope(approved) if replay else steps_scope(approved, steps) if steps else None) reopen = tuple(_replay["reopen"]) if _replay else () if _replay: approved = _replay["approved"] rid = run_id or cbs.new_run_id() root = ci.root_dir(rid) # ★★재개 판이면 이번 attempt 의 검색·받기 문을 **다시 센다** (Codex BLOCK). # ★코드가 다시 여는 스텝(지문 force · 빚)은 계획을 지은 뒤에야 알므로 그때 **한 번 더** 센다. approved_table = dict(approved) approved = outbound_doors_on_resume(rid, approved_table, reopen=reopen) root.mkdir(parents=True, exist_ok=True) # ★①격리 — **무엇보다 먼저.** 여기서 `DATABASE_URL` 을 갈아 둔다. # ★`SessionLocal` 을 실제로 쓰는 **유료 주행에만** 이 문을 건다 — # dry 는 DB 를 안 쓰므로 막을 것이 없다(시험 프로세스에는 conftest 가 # 이미 올려 둔다). if live: ci.assert_database_module_not_loaded() env = cbs.prepare_env(rid, grounding_mode=sc["mode"]) os.environ.update(env) # ★★★**첫 저장·DB·provider 보다 앞에** 선다 — 이어 갈 자격이 있나. # `record_code_transition` 을 만들어 놓고 **아무도 안 불렀다** # (Codex BLOCK 2026-09-02: 호출자 0). 이제 여기가 그 문이다. resumed = assert_resume_transition(rid, sc) if resumed.get("resume"): resumed["setting_flip"] = assert_setting_flip_is_safe(rid, sc) # ★hash adoption 은 live 전에 **이미 적용된 사건**이다(`canary_hash_adoption.adopt`) — 여기서는 전이가 결속한 사건 id 만 옮겨 적는다. _adoptions = list((resumed or {}).get("hash_adoptions") or []) got: Dict[str, Any] = {"run_id": rid, "live": live, "scenario": dict(sc), "code": git_tip(), "replay": _replay, "resume": resumed, "started_kst": datetime.now(KST).isoformat( timespec="seconds"), "stages": {"isolation": { "db": ci.db_name(rid), "projects_dir": env["PROJECTS_DIR"]}}} #: 이 판의 산출 파일. ★첫 판은 `canary_run.json`, 재개는 **따로**. out_name = "canary_run.json" if not resumed["resume"] else ( f"canary_resume_{rid}_{attempt_seq(root)}.json") def _save() -> None: """★★첫 판의 `canary_run.json` 을 **안 덮는다** (Codex BLOCK 09-02). 앞 판은 재개할 때마다 그 파일을 새로 써서 **첫 판의 stages 가 사라졌다** — 무엇으로 얼마 썼는지가 증거인데 그것이 없어진다. 재개는 제 파일을 따로 쓰고, 첫 판은 그대로 둔다. """ (root / out_name).write_text( json.dumps(got, ensure_ascii=False, indent=1, default=str), encoding="utf-8") got["output_file"] = out_name _save() # ★여기서부터 `app.core.*` 가 올라와도 **이미 canary 를 가리킨다** built = build_plan(sc, run_id=run_id) # ★★★단위를 안 읽은 스텝이 **살 자리**에 있으면 provider 전에 선다 # (Codex 재리뷰 2026-09-02: single 로 접지 않는다). 끝난 스텝은 cap 0 이라 # 살 자리가 아니다 — 그것까지 막으면 dry 가 못 돈다. _done0 = set(completed_steps_of(rid, statuses=("completed", "not_applicable"))) # ★★끝났어도 지문이 어긋나 force 될 스텝은 **산다** — 남은 표·상한 문·단위 확인에서 done 으로 치지 않는다 # (실측 4398a55dc0bb 재개: 야외 스텝이 completed 인 채 force 였는데 계획에서 빠졌다) # ★계측 스텝만이 아니라 **적용 스텝 전부** — 무료 스텝이 지문 어긋남으로 force 되면 하류(유료)가 stale 로 다시 구워진다 # (실측 f7cc45c576c0 plain resume: episode_reference_policy 가 매번 force → scene_detail 15~19콜 재구매, dry 는 forced [] 라 했다) _applied_all = {(r["step"] if isinstance(r, dict) else str(r)) for r in (built["plan"].get("applied") or [])} _dr = drift_reentry_of(rid, sorted(_done0 & _applied_all), sc) got["forced_for_drift"] = _dr["forced"] got["hash_adoptions"] = _adoptions # ★★force 가 지울 하류 — dry 가 이것을 안 보여 주면 「살 자리 5」가 실제로는 「run 전체 다시」다 got["forced_would_invalidate"] = forced_would_invalidate_of(rid, _dr["forced"]) got["reopened_for_debt"] = reopened_for_debt_of( rid, sorted(_done0 & set(built["plan"]["applied_metered"])), sc) _done0 = _done0 - set(got["forced_for_drift"]) - set(got["reopened_for_debt"]) # ★코드가 다시 연 스텝까지 합쳐 검색·받기 문을 다시 센다 — 빚으로 다시 여는 중앙은 검색을 산다 _reopened_all = set(reopen or ()) | set(got["forced_for_drift"]) | set(got["reopened_for_debt"]) _applied_ids = [(r["step"] if isinstance(r, dict) else str(r)) for r in (built["plan"].get("applied") or [])] approved = outbound_doors_on_resume(rid, approved_table, reopen=sorted(_reopened_all), applied=_applied_ids) got["outbound_doors_this_attempt"] = {"search": approved["search"], "download": approved["download"], "reopened": sorted(_reopened_all), "consumers": approved.get("★outbound_consumers"), "why": approved.get("★doors_note")} _unv = [x for x in built["unverified_units"] if x not in _done0] got["plan_totals"] = dict(built["totals"]) got["plan_rows_remaining"] = [ {"step": r["step"], "unit": metering_unit_of(r["step"]), "normal": r["normal_expected"], "hard_cap": int(r["logical_cap"]), "calls_per_unit": r["calls_per_unit"]} for r in built["rows"] if r["step"] not in _done0] _gate_dims = {**built["dimensions"], **(built.get("plan_basis") or {})} # 문에는 치수 + 계획 근거 got["entity_cap_check"] = assert_entity_cap_covers_queue( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) got["outlook_pair_cap_check"] = assert_outlook_pair_cap_covers( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) got["state_variant_cap_check"] = assert_state_variant_cap_covers( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) got["outdoor_group_cap_check"] = assert_outdoor_group_cap_covers( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) got["central_cap_check"] = assert_central_cap_covers( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) got["chain_group_cap_check"] = assert_chain_group_cap_covers( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) got["floor_plan_cap_check"] = assert_floor_plan_cap_covers( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) got["background_cap_check"] = assert_background_cap_covers( rid, _gate_dims, _done0, built["plan"]["applied_metered"]) # ★★이미지 문의 남은 표 — 승인(APPROVED_IMAGE_CALLS)은 **정상 합**과 견준다. # 0 이면 문 앞에서 서던 것을 **시작 전에** 세운다 (글 문과 같은 규칙). _img_rows = [r for r in built["images"]["rows"] if r["step"] not in _done0] _exp = expected_calls_of(rid) for r in _img_rows: r["expected_units"] = _exp["by_step"].get(r["step"], {}).get("units", 0) r["expected_images"] = _exp["by_step"].get(r["step"], {}).get("images", 0) r["expected_text"] = _exp["by_step"].get(r["step"], {}).get("text", 0) got["image_plan_remaining"] = { "rows": _img_rows, "normal_total": int(sum(r["normal"] for r in _img_rows)), "hard_total": int(sum(r["hard"] for r in _img_rows)), # ★CP 에서 읽은 **실제로 살 수** — 승인은 이 수 + 명시 여유 (Codex 2026-09-02) "expected_images": int(sum(r["expected_images"] for r in _img_rows)), "expected_text": int(sum(r["expected_text"] for r in _img_rows)), "approved": approved_images_for(sc)} # ★이미지를 사는 스텝은 이 범위의 허용 집합 안에만 — 다른 source 가 보이면 provider 전에 선다 (dry 도 같이 본다) got["image_sources"] = assert_image_sources(built, sc) if live and got["image_plan_remaining"]["expected_images"] > approved_images_for(sc): raise ci.IsolationRefused( f"이미지를 한 번씩만 성공해도 {got['image_plan_remaining']['expected_images']} 장인데 " f"이 범위의 승인은 {approved_images_for(sc)} 이다 — 사람이 다시 정해야 한다") got["plan_unverified_units_live"] = _unv if live and _unv: raise ci.IsolationRefused( f"계측 단위를 안 읽은 스텝 {_unv} 이 살 자리에 있다 — 읽어서 " "METERING_UNITS 에 적기 전엔 안 산다") # ★★계획과 실행이 **같은 시나리오**인가 — 세 축 중 하나만 달라도 선다 got["stages"]["scope"] = assert_scope(built, sc) # ★★좁힌 판이면 **도는 스텝만의** 수를 따로 낸다 (Codex 2026-09-02). # closure 전체 수(69)를 보이면 이 판이 실제로 무엇을 사는지가 안 읽힌다. # ★수는 `build_plan` 의 같은 `caps`·`rows` 에서 **잘라** 낸다 — 다시 # 세지 않는다. 문(`canary_pipeline`)도 같은 `caps` 를 쓴다. narrowed = None if reopen: want = [r for r in built["rows"] if r["step"] in set(reopen)] narrowed = { "steps": list(reopen), "caps": {s: built["caps"].get(s) for s in reopen}, "rows": [{k: r.get(k) for k in ("step", "model_physical", "logical_cap", "counted_cap", "raw_upper", "key_slots", "slots_how")} for r in want], "logical_total": sum(int(r["logical_cap"]) for r in want), "counted_total": sum(int(r["counted_cap"]) for r in want), "raw_total": sum(int(r["raw_upper"]) for r in want), } got["plan"] = { "narrowed": narrowed, "closure": built["plan"]["closure_total"], "applied": len(built["plan"]["applied"]), "metered": len(built["plan"]["applied_metered"]), "free": built["plan"]["applied_free"], "unknown": built["plan"]["applied_unknown"], "skipped": [r["step"] for r in built["plan"]["skipped"]], "fixture_dimensions": built["dimensions"], # ★★이름을 바로잡는다 (Codex 조건 ⑦) — 이 칸은 **논리 수**다. # `expected_counted` 라 적혀 있어서 내가 「세는 물리」로 읽고 # 「승인 밖」이라 잘못 보고했다. 승인선과 견주는 것은 **이 수**다. "planning_logical": built["totals"]["logical_cap_total"], "worst_case_counted": built["totals"]["counted_cap_total"], "worst_case_raw": built["totals"]["raw_upper_total"], "emergency_counted": approved["counted"], "emergency_raw": approved["raw"], # ★★셋은 **서로 다른 문**이다 — 글 예산이 검색도 받기도 못 센다 "approved_search_requests": approved["search"], "approved_download_operations": approved["download"], "approved_image_calls": approved_images_for(sc), "approved_images_scope": list(scope_key(sc)), # ★partial 스텝 — cap 은 그대로 열리고 기존 완료분은 다시 안 산다 "partial_steps": partial_steps_of(rid, built["caps"]), "★image_note": ( "글 예산은 이미지 문을 **못 본다**. 이 closure 안에서 유료 " "이미지를 사는 것은 `floor_plan_render` 하나이고, 이 판의 " "이미지 승인은 위 수다 — 0 이면 문 앞에서 선다"), "★ceiling_note": ( "문에 쓰는 것은 **승인값**이다. `expected` 는 한 번씩 성공했을 " "때의 수(논리와 같다)이고, `worst_case` 는 tier·슬롯이 다 열렸을 " "때다. ★승인 상한은 worst_case **아래**로 일부러 둔다 — 실패가 " "쏟아지면 **일찍 멈추라고** 두는 정지선이지 상한 예측이 아니다."), "bootstrap_cap": BOOTSTRAP_CAP, # ★잠근 계약 — dry 에서도 **보인다**. live 에서만 보이면 승인 전에 # 무엇이 잠겼는지 못 읽는다 (Codex 조건 ⑤). "locks": dict(LOCKED_CONTRACT), } _save() # ★★계산값이 **승인 밖**이면 선다. 승인은 내가 센 수 **위에** 있다. # ★★재는 것은 **expected** 다 — 한 번씩 성공했을 때의 수가 승인 안이어야 # 한다. worst_case 는 그 위여도 된다(정지선이 먼저 선다). exp = built["totals"]["logical_cap_total"] if exp is None or int(exp) > approved["counted"]: raise ci.IsolationRefused( f"한 번씩만 성공해도 {exp} 인데 승인은 " f"{approved['counted']} 이다 — 사람이 다시 정해야 한다") if built["plan"]["applied_unknown"]: raise ci.IsolationRefused( f"유료 여부가 미확정인 스텝이 있다: " f"{built['plan']['applied_unknown']} — 정하기 전에 안 돈다") if not live: got["note"] = "★안 샀다 — `--live` 여야 산다" _save() return got # ★②DB got["stages"]["database"] = ci.create_database(rid) _save() # ★③schema — **별도 프로세스** child = ci.child_env(rid, env=env) proc = subprocess.run(ci.upgrade_command(rid), env=child, cwd=str(BACKEND), capture_output=True, text=True) # ★★alembic 은 접속 주소를 찍을 수 있다 — **가려서** 저장한다 got["stages"]["alembic"] = { "returncode": proc.returncode, "stdout": ci.masked_text(proc.stdout[-2000:]), "stderr": ci.masked_text(proc.stderr[-2000:])} _save() if proc.returncode != 0: raise ci.IsolationRefused( f"schema 를 못 올렸다 (코드 {proc.returncode}) — 여기서 끝낸다") # ★★요청 계약을 **먼저** 잠근다 — 부트스트랩의 유료 1회도 그 안이다. # ★부트스트랩이 Router 를 **짓는다**(`create_project` → 이름 저작). # 그 뒤에 잠그려 하면 「이미 지어졌다」로 서므로 **한 번에 감싼다**. from tools.grounding_audit import canary_request_lock as rl # ★★★운반층 상한 — 스텝이 제 예산 scope 를 열어도 **가려지지 않는다**. # 실측 2026-09-02: `grounding_chunk` 이 `research_run_scope(cap=24)` 로 # canary 예산을 갈아 끼워 그 스텝의 유료 2건이 **counted 0** 으로 적혔다. # 즉 「run 전체 정지선」이 그 스텝을 못 봤다. 이제 여기서 다 센다. # ★★★**93 을 새로 열지 않는다** (Codex 2026-09-02). 이 run 이 이미 쓴 # 유효 누계를 빼고 남은 것만 연다 — 안 그러면 총 노출이 93 을 넘는다. _left = cp.remaining_cap(root, ceiling=int(approved["counted"])) got["stages"]["transport_budget"] = { "ceiling": int(approved["counted"]), "cumulative_before": cp.cumulative_used(root), "opened": _left, "★means": ("운반층 상한은 **남은 것**이다 — 승인선을 판마다 새로 " "열면 run 전체 상한이 아니게 된다")} _save() _tpath = root / cp.TRANSPORT_LOG #: 지금 어느 attempt 의 몫인가. ★pipeline 이 열리면 그 id 로 바뀐다. _aid = {"id": ""} def _sink(snap): """★★전송 **직전마다** 줄을 **덧붙인다** — `os._exit` 여도 남는다. ★★★덮어쓰면 안 된다 (Codex 2026-09-02): 앞 판이 죽어 남긴 수를 다음 판이 0 으로 **지운다**. 그래서 append-only 줄이고, 각 줄에 run·attempt·scope 신원을 함께 적는다. ★attempt 줄이 아니라 **파일**이다 — attempt 는 pipeline 안에서 열리는데 부트스트랩의 전송은 그 전에 난다. """ row = {"run_id": rid, "pid": os.getpid(), # ★★줄마다 **attempt 신원**을 결속한다 — 안 그러면 「장부에 # 이미 접힌 것」과 「아직 미정인 것」을 못 가른다 "attempt_id": _aid["id"], "at": datetime.now(KST).isoformat(timespec="seconds"), **snap} with open(_tpath, "a", encoding="utf-8") as fh: fh.write(json.dumps(row, ensure_ascii=False, default=str) + "\n") fh.flush() os.fsync(fh.fileno()) # ★죽어도 남게 def _transport_so_far(scope: str): """**아직 장부에 안 접힌** 몫만. ★두 번 차감하지 않는다.""" got = cp.unsettled_transport(root, scope) return int(got["used"]), dict(got["by_source"]) # ★부트스트랩 몫으로 **먼저** 연다 — pipeline 과 상한이 다르다 with rl.canary_request_lock( transport_cap=BOOTSTRAP_CAP, transport_name="bootstrap", transport_sink=_sink) as locked: # ★★앞 판이 죽어 남긴 수를 **이어받는다** — 0 으로 안 지운다 _b_used, _b_by = _transport_so_far("bootstrap") locked["transport"].adopt(_b_used, _b_by) got["stages"]["request_lock"] = { k: v for k, v in locked.items() if k != "transport"} _transport = locked["transport"] # ★★**첫 유료 호출 전에** Router 를 짓고 **즉시** 확인한다. # 아무것도 안 보낸다 — 객체만 짓고 `num_retries` 를 읽는다. # 0 이 아니거나 못 읽으면 여기서 **provider 0 으로** 선다. got["stages"]["router_ready"] = rl.prepare_router() _save() # ★④부트스트랩 — **제 상한** (예산은 따로, 계약 잠금은 같이) # ★★★**실제 엔진**이 canary DB 인지 — env 가 아니라 붙은 URL 을 본다 got["stages"]["engine_before_bootstrap"] = ci.assert_engine_is_canary(rid) boot = cbs.bootstrap(rid, live=True, cap=BOOTSTRAP_CAP, fixture=sc["fixture"]) got["stages"]["bootstrap"] = boot got["stages"]["router_lock_after_bootstrap"] = \ rl.assert_router_locked() _save() # ★⑤pipeline — **새 예산**. 운반 몫도 여기서 갈아 낀다. # ★★★신원을 **여기서 만들어** pipeline 과 운반 기록이 **같은 것**을 # 받게 한다. 앞 판은 `got["stages"]["pipeline"]` 을 먼저 읽었는데 # 그 자리는 `run_pipeline` **뒤에** 생기므로 늘 빈 문자열이었다. import uuid as _uuid _aid["id"] = _uuid.uuid4().hex[:12] locked["transport"].set_scope("pipeline", _left) _p_used, _p_by = _transport_so_far("pipeline") locked["transport"].adopt(_p_used, _p_by) got["stages"]["transport_reconciled"] = { "adopted": _p_used, "by_source": _p_by, "★means": ("장부에 **아직 안 접힌** 운반만 이어받는다 — " "terminal 로 닫힌 attempt 의 몫은 이미 누계에 있다")} got["stages"]["engine_before_pipeline"] = ci.assert_engine_is_canary(rid) got["stages"]["pipeline"] = cp.run_pipeline( run_id=rid, project_id=boot["project_id"], episode_id=boot["episode_id"], plan=built["plan"], caps=built["caps"], attempt_id=_aid["id"], emergency_counted=approved["counted"], # ★★중앙이 **첫 outbound 전에** 다섯 갈래를 확인한다 — # 다르면 다른 것을 재는 것이라 provider 0 으로 선다 project_config=_owner_requirement(sc["fixture"]), approved_search=approved["search"], approved_downloads=approved["download"], approved_image_calls=approved_images_for(sc), reopen=reopen, live=True, # ★producer 상한 문 — 그 단위의 스텝을 부르기 직전에 **지금 있는 CP** 로 센다 before_step=lambda s: producer_cap_gate_before( s, run_id=rid, dims=_gate_dims, done=_done0, metered=built["plan"]["applied_metered"])) # ★★★**산 것을 판정보다 먼저 저장한다** (실측 2026-09-02 유료 canary ①). # 앞 판은 부트스트랩 뒤로 `_save()` 가 **한 번도 없었다**. 그래서 # 파이프라인이 다 돈 뒤 `assert_client_retries_observed()` 가 서자 # 마지막 `_save()` 에 못 닿았고, **유료 주행의 pipeline 단계가 이 # 판의 산출 파일에서 통째로 사라졌다**(운반·문 셋·스텝별 수 전부). # `pipeline_run.json` 이 따로 남아 살았을 뿐이다. _save() got["stages"]["transport"] = _transport.snapshot() _save() got["stages"]["router_lock"] = rl.assert_router_locked() _save() got["stages"]["client_retries"] = rl.assert_client_retries_observed() _save() got["finished_kst"] = datetime.now(KST).isoformat(timespec="seconds") _save() return got def main() -> int: """★★세 축을 **명시로** 받는다 — 조용히 legacy/scene_detail 로 안 떨어진다. canary_run.py --dry --mode v2_chunk --fixture period_episode \ --target reference_acquisition """ import argparse ap = argparse.ArgumentParser(add_help=True) g = ap.add_mutually_exclusive_group(required=True) g.add_argument("--dry", action="store_true") g.add_argument("--live", action="store_true") ap.add_argument("--replay", action="store_true", help=("받아 둔 사진만 **다시 판정**한다 — 검색·받기 문을 " "0 으로 박고 중앙 스텝을 다시 연다")) ap.add_argument("--steps", default="", help=("쉼표로 적은 **그 스텝만** 돈다 — 검색·받기 문을 0 으로 " "박는다. closure 전부를 안 돈다")) ap.add_argument("--run-id", default=None) # ★기본값을 여기 안 적는다 — `scenario()` 한 곳이 안다 ap.add_argument("--mode", default=DEFAULT_CANARY_MODE) ap.add_argument("--fixture", default=cbs.DEFAULT_FIXTURE) ap.add_argument("--target", default="scene_detail") ap.add_argument("--background", default=None, choices=("on", "off"), help="배경 모드 **선언** — process 의 실제 설정과 다르면 선다") ap.add_argument("--still-recipe", default=None, choices=("on", "off"), help="still_recipe 사슬 **선언** — `STILL_RECIPE_MODE=off` 로 끈다") ap.add_argument("--outdoor", default=None, choices=("on", "off"), help="outdoor 사슬 **선언** — OUTDOOR_LANE_*_ENABLED=false 로 끈다") a = ap.parse_args() # ★★★이 판의 env 를 **settings 가 올라오기 전에** 박는다 (실측 2026-09-02 # 격리 near-miss): `scenario()` 가 배경 술어를 물으며 `app.core.config` 를 # 올렸고, 그 뒤 `run()` 이 env 를 갈아도 settings 는 이미 .env 의 원본 # DB URL 로 굳어 있었다. 그래서 뒤에 올라온 `app.core.database` 가 # **원본에 붙었고** 부트스트랩이 canary 프로젝트를 못 찾아 섰다 — # 그 거절이 없었으면 pipeline 이 원본 DB 로 돌 뻔했다. ct.assert_backend_cwd() rid = a.run_id or cbs.new_run_id() os.environ.update(cbs.prepare_env(rid, grounding_mode=a.mode)) # ★★★fail-closed — env 가 박힌 **이 순간** settings/database 가 아직 안 # 올라와 있어야 한다. 이미 올라와 있으면 env 를 바꿔도 엔진은 안 바뀐다 # (Codex 조건 2 · 실측: settings 가 원본 URL 로 굳은 채 database 가 붙었다). ci.assert_app_modules_not_loaded() sc = scenario(mode=a.mode, fixture=a.fixture, target=a.target, background=a.background, still_recipe=a.still_recipe, outdoor=a.outdoor) got = run(live=a.live, run_id=rid, sc=sc, replay=a.replay, steps=[x.strip() for x in str(a.steps).split(",") if x.strip()]) print(f"■ canary {got['run_id']} · {'유료' if got['live'] else 'dry'}") p = got["plan"] print(f" 적용 {p['applied']} (계측 {p['metered']} · 무료 " f"{len(p['free'])} · 미확정 {len(p['unknown'])}) · 건너뜀 " f"{len(p['skipped'])}") d = p["fixture_dimensions"] print(f" 원고 {d['scenes']}씬 {d['shots']}샷 · fan_out 상한 " f"{d['fan_out_cap']}") print(f" 범위 {got['scenario']['mode']} · {got['scenario']['fixture']} " f"· 배경 {got['scenario'].get('background')} " f"· still_recipe {got['scenario'].get('still_recipe')} " f"· outdoor {got['scenario'].get('outdoor')} " f"→ {got['scenario']['target']} · 코드 " f"{got['code']['tip'][:8]}{'' if got['code']['clean'] else ' ★더러움'}") if got.get("plan_unverified_units_live"): print(f" ★단위 미확인(살 자리): {got['plan_unverified_units_live']} — live 는 여기서 선다") if got.get("hash_adoptions"): print(f" ★hash adoption 결속됨(전이): {got['hash_adoptions']} — data 불변 · config_hash 메타데이터만 · 그 스텝은 cap 0 으로 되쓴다") for _s, _lst in (got.get("forced_would_invalidate") or {}).items(): print(f" ★★지문 어긋남 · force({_s}) — 하류 {len(_lst)} CP 무효화(삭제): {_lst[:12]}{' …' if len(_lst) > 12 else ''}") for _s, _v in (p.get("partial_steps") or {}).items(): print(f" ★partial {_s} — cap {_v['cap']} · 기존 완료 {_v['completed']} " f"재구매 0 · 예상 신규 {_v['expected_new']} (applicable {_v['applicable']})") print(f" 문 셋 — 검색 요청 {p['approved_search_requests']} · 받는 장수 " f"{p['approved_download_operations']} · 이미지 생성 " f"{p['approved_image_calls']}") if got.get("replay"): _ap = (got.get("replay") or {}).get("approved") or {} print(f" ★좁힌 판 — 도는 스텝 " f"{list(got['replay']['reopen'])} · 검색 문 {_ap.get('search')} · 받기 문 {_ap.get('download')}") nw = p.get("narrowed") or {} for r in nw.get("rows") or (): print(f" {r['step']:18} 논리 {r['logical_cap']} · counted " f"{r['counted_cap']} · raw {r['raw_upper']} · 슬롯 " f"{r['key_slots']}({r['slots_how']})") print(f" 합계 논리 {nw.get('logical_total')} · 슬롯 최악 " f"{nw.get('counted_total')} · raw {nw.get('raw_total')}") print(f" 잠금 {p['locks']}") for _r in got.get("plan_rows_remaining") or []: print(f" ★살 자리 {_r['step']:26} {_r['unit']:20} 정상 {_r['normal']:>2} " f"· hard cap {_r['hard_cap']:>3} (×{_r['calls_per_unit']})") _ip = got.get("image_plan_remaining") or {} for _r in _ip.get("rows") or []: print(f" ★이미지 {_r['step']:26} {_r['unit']:14} 단위≤{_r['units_cap']:>2} " f"(실제 {_r.get('expected_units', 0):>2}) 정상 {_r['normal']:>3} hard {_r['hard']:>4} " f"· 실제 이미지 {_r.get('expected_images', 0):>3} 글 {_r.get('expected_text', 0):>3}") if _ip: print(f" ★이미지 합: 상한 정상 {_ip['normal_total']} · hard {_ip['hard_total']} " f"· **실제 이미지 {_ip['expected_images']} · 글 {_ip['expected_text']}** " f"· 승인 {_ip['approved']}") _hard = (got.get("plan_totals") or {}).get("hard_cap_total") if _hard is not None: print(f" ★정상 예상 합 {p['planning_logical']} · 스텝 hard cap 합 {_hard}") print(f" 계획 논리 {p['planning_logical']} · 최악 " f"{p['worst_case_counted']}/raw {p['worst_case_raw']} · " f"★승인 정지선 {p['emergency_counted']}/{p['emergency_raw']} · " f"부트스트랩 {p['bootstrap_cap']}") # ★★재개면 **이 run 이 이미 쓴 것**과 남은 것을 함께 적는다 — # 승인선만 적으면 판마다 새로 여는 것처럼 읽힌다 (Codex 09-02). from tools.grounding_audit import canary_pipeline as _cp _root = ci.root_dir(got["run_id"]) _res = got.get("resume") or {} if _res.get("resume"): _cum = _cp.cumulative_used(_root) _left = _cp.remaining_cap(_root, ceiling=int(p["emergency_counted"])) print(f" 재개 {_res.get('from_tip','')[:8]}→" f"{_res.get('to_tip','')[:8]} · 누계 {_cum} · 남은 {_left}" f" · 열린 attempt {len(_cp.open_attempts(_root))}") # ★적은 곳을 **실제로 쓴 이름**으로 적는다. 앞 판은 재개일 때도 # `canary_run.json` 이라 적어, 첫 판 증거를 덮은 것처럼 읽혔다. print(f" 적었다: {_root / got['output_file']}") return 0 if __name__ == "__main__": raise SystemExit(main())