"""W19E8 — bounded real-API opt-in surface (no real run yet). Builds on W19E7 fake-wrapper smoke. Adds dual gate (--allow-real-api CLI + W19E8_ALLOW_REAL_API=1 env), real-client lazy resolvers (still fail-closed in this wave — execution gated until W19E9 Codex re-review), hard per-counter caps, and an output-artifact audit (sha256/size). Default --generate path stays fake-safe (W19E7 contract preserved). No external LLM/image/VLM calls, no DB writes, no ImageAsset writes, no commit/push. ----- W19E7 baseline (still applies) ----- W19E7 — narrow opt-in production smoke (fake-generate wiring). Scope: - Read existing checkpoints for a single fp + selected bg_ids. - Emit dry-run run_meta.json + index.html with planned API/image counts, expected output paths, target fp/bg existence verification, and a background-gate provenance summary (chain_bg/indoor/shot_count≥3 from background_classify cp; (loc_id, sub_location) link from background_master_plan cp). - ``--generate`` is real-API-safe: the 5 default seams still raise ``RuntimeError`` (W19E6 contract preserved). Each seam has a paired ``_wrap_*`` helper that drives the **actual production wrapper** (``build_fp_user_prompt`` / ``run_floor_plan_prompt`` / ``render_one_floor_plan`` / ``build_overlay_payload`` / ``build_bg_user_prompt`` / ``run_background_prompt`` / ``render_one_background``). Tests monkeypatch each seam to delegate to its ``_wrap_*`` with an injected fake ``call_structured_fn`` and fake OpenAI image client → API 0, DB 0, ImageAsset 0. Out of scope (future narrow patch): - Real OpenAI/LLM gate. Wrappers refuse client construction; the real-API path must be wired explicitly in a later wave. Guards: - ``_check_clean_or_approved_w19_diff()`` accepts diffs only inside the W19A-D approved scope (``backend/alembic`` always fails). The earlier fully-empty ``backend/app`` check is replaced because W19A-D wave is intentionally untracked. - No DB write. No ImageAsset write. No real API call (fake-generate exercises wrappers but never reaches a real client). - Failure modes: missing checkpoint, missing target fp/bg, group not chain_bg, or diff violations outside approved paths → exit_code=1 with explicit ``failed_invariants``. """ from __future__ import annotations import argparse import hashlib import html import json import os import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple _REPO_ROOT = Path(__file__).resolve().parents[2] _SCRIPTS_DIR = _REPO_ROOT / "backend" / "scripts" if str(_SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(_SCRIPTS_DIR)) import subprocess # noqa: E402 from experiment_background_pipeline_slice import ( # type: ignore _check_production_diff_empty, ) # W19E7 (Codex narrow patch): approved diff scope. Files outside this set, # any ``backend/alembic`` diff, or any DB schema touch must fail the gate. # Listed verbatim from the W19A-D wave gitStatus snapshot. _W19E7_APPROVED_DIFF_PATHS: frozenset = frozenset({ "backend/app/core/config.py", "backend/app/core/step_manifest.py", "backend/app/core/steps/__init__.py", "backend/app/core/steps/background_prompt_step.py", "backend/app/core/steps/background_render_step.py", "backend/app/core/steps/floor_plan_overlay_payload_step.py", "backend/app/core/steps/floor_plan_prompt_step.py", "backend/app/modules/pipeline/background_image_planner.py", "backend/app/modules/pipeline/background_prompt.py", "backend/app/modules/pipeline/floor_plan_overlay_payload.py", "backend/app/modules/pipeline/floor_plan_prompt.py", "backend/app/modules/prompt_loader.py", }) # W19E7-R1 G2: approved prompt-pack directory prefixes. Any tracked or # untracked path under one of these prefixes (W19A v6 floor_plan_prompt pack, # W19B-2 v7 background_prompt pack) is in scope; any other ``prompts/_base/`` # touch is a violation so an unexpected prompt drift cannot silently pass # this smoke. _W19E7_APPROVED_PROMPT_PACK_PREFIXES: tuple = ( "prompts/_base/floor_plan_prompt/6.202605262300/", "prompts/_base/background_prompt/7.202605262330/", ) _PROMPT_AUDIT_ROOT: str = "prompts/_base" def _collect_diff_paths(scope_paths: List[str]) -> Tuple[List[str], Optional[str]]: """Run ``git diff --name-only`` + ``git status --porcelain`` against the given scope and return the unique, sorted set of touched paths. Returns ``(paths, error_message)``. On a git failure ``paths`` may be empty and ``error_message`` carries the truncated reason for the caller to fold into violations. """ diff_names: List[str] = [] err: Optional[str] = None try: r = subprocess.run( ["git", "diff", "--name-only", *scope_paths], cwd=str(_REPO_ROOT), capture_output=True, text=True, check=False, ) diff_names = [ line.strip() for line in r.stdout.splitlines() if line.strip() ] except Exception as exc: return [], f"git_diff_failed: {exc}"[:200] try: s = subprocess.run( ["git", "status", "--porcelain", *scope_paths], cwd=str(_REPO_ROOT), capture_output=True, text=True, check=False, ) for line in s.stdout.splitlines(): line = line.rstrip() if len(line) < 4: continue path = line[3:].strip() if path and path not in diff_names: diff_names.append(path) except Exception as exc: err = f"git_status_failed: {exc}"[:200] return diff_names, err def _expand_dir_marker(path: str, *, scope: str) -> List[str]: """Resolve a directory porcelain entry ('?? prompts/_base/foo/') to the concrete file paths under it so the audit can match approved-prefix rules against real files instead of bare dir markers. """ if not path.endswith("/"): return [path] abs_dir = _REPO_ROOT / path if not abs_dir.is_dir(): return [path] expanded: List[str] = [] for p in sorted(abs_dir.rglob("*")): if p.is_file(): try: rel = p.relative_to(_REPO_ROOT) except ValueError: continue expanded.append(str(rel)) return expanded or [path] def _check_clean_or_approved_w19_diff() -> Dict[str, Any]: """W19E7-R1 G2 production_diff guard — accept approved W19A-D scope only. Two audits combined: - ``backend_app_alembic_guard``: ``backend/app`` and ``backend/alembic`` scope. Approved exact-path set (``_W19E7_APPROVED_DIFF_PATHS``); ``backend/alembic`` always fails. - ``prompt_pack_audit``: ``prompts/_base`` scope. Approved prefix set (``_W19E7_APPROVED_PROMPT_PACK_PREFIXES``); any prompt path outside those packs is a violation so an unexpected prompt change cannot silently pass this smoke. Returns a dict the caller can fold into ``run_meta`` and decide fail/pass on ``is_clean_or_approved``. Pure read-only. """ backend_paths, backend_err = _collect_diff_paths( ["backend/app", "backend/alembic"] ) prompt_paths_raw, prompt_err = _collect_diff_paths([_PROMPT_AUDIT_ROOT]) prompt_paths: List[str] = [] for p in prompt_paths_raw: prompt_paths.extend(_expand_dir_marker(p, scope=_PROMPT_AUDIT_ROOT)) violations: List[str] = [] alembic_diff: List[str] = [] approved_seen: List[str] = [] prompt_violations: List[str] = [] prompt_approved_seen: List[str] = [] if backend_err: violations.append(backend_err) if prompt_err: violations.append(prompt_err) for path in backend_paths: if path.startswith("backend/alembic"): alembic_diff.append(path) continue if path in _W19E7_APPROVED_DIFF_PATHS: approved_seen.append(path) continue violations.append(path) for path in prompt_paths: if any( path.startswith(prefix) for prefix in _W19E7_APPROVED_PROMPT_PACK_PREFIXES ): prompt_approved_seen.append(path) continue prompt_violations.append(path) is_clean_or_approved = ( not violations and not alembic_diff and not prompt_violations ) return { "is_clean_or_approved": is_clean_or_approved, "violations": sorted(violations), "alembic_violations": sorted(alembic_diff), "approved_paths_seen": sorted(approved_seen), "all_diff_paths": sorted(backend_paths), "prompt_pack_audit": { "violations": sorted(prompt_violations), "approved_paths_seen": sorted(prompt_approved_seen), "all_diff_paths": sorted(prompt_paths), "approved_prefixes": list(_W19E7_APPROVED_PROMPT_PACK_PREFIXES), }, } KST = timezone(timedelta(hours=9)) PLAN_VERSION = "w19e8_v1" STAGE = "w19e_production_opt_in_single_fp_smoke" # W19E8 hard caps for the 1-fp / 3-bg smoke. No retry margin: production # helpers configured with ``max_retries=1`` where possible, and the cap # enforcer aborts immediately on any breach (no further calls attempted). # Counters in run_meta record successful exterior work — caps gate the # attempt rate, not the success rate. W19E8_API_CALL_CAPS: Dict[str, int] = { "llm_api_call_count": 4, "image_api_call_count": 4, "vlm_api_call_count": 0, "db_write_count": 0, "image_asset_write_count": 0, } W19E8_REAL_API_ENV_VAR = "W19E8_ALLOW_REAL_API" W19E8_REAL_API_ENV_VALUE = "1" DEFAULT_PROJECT_ID = "8d56bc5d-89eb-4733-9890-cbec35dd358b" DEFAULT_EPISODE_ID = "1458fcc5-fb7d-407c-aa45-bc41bf98bba7" DEFAULT_FP_ID = "fp_police_office_main" DEFAULT_BG_IDS = "L14B01,L14B02,L14B03" DEFAULT_OUTPUT_ROOT = ( _REPO_ROOT / "scripts_output" / "w19e_production_opt_in_single_fp_smoke" ) class _ApiCallCapExceeded(RuntimeError): """Hard cap breach — abort run before any further external attempt.""" def _enforce_cap_or_raise( *, counter_name: str, counters: Dict[str, int], caps: Dict[str, int] ) -> None: """Pre-call cap check. Raises ``_ApiCallCapExceeded`` if incrementing the given counter would exceed its W19E8 cap. Counter is NOT incremented here — caller does that AFTER the underlying call returns successfully. """ planned = counters[counter_name] + 1 cap = caps[counter_name] if planned > cap: raise _ApiCallCapExceeded( f"api_call_cap_exhausted: {counter_name} would reach {planned} " f"(cap={cap}); aborting before any further external attempt" ) def _real_api_gate( *, cli_allow_real_api: bool, env: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """Dual-gate evaluation. Real-API mode requires BOTH the CLI flag AND the env var set to ``"1"``. Default path stays fake-safe. ``real_api_gate_status`` taxonomy: - ``disabled_default`` neither set - ``disabled_env_only`` env set, no CLI flag - ``disabled_flag_only`` CLI flag set, no env - ``enabled`` both set (real-API path entered, but W19E8 resolvers still fail-closed) """ src = env if env is not None else os.environ env_ok = src.get(W19E8_REAL_API_ENV_VAR) == W19E8_REAL_API_ENV_VALUE if cli_allow_real_api and env_ok: status = "enabled" elif cli_allow_real_api and not env_ok: status = "disabled_env_only" # flag without env → still disabled elif env_ok and not cli_allow_real_api: status = "disabled_flag_only" # env without flag → still disabled else: status = "disabled_default" return { "real_api_mode": cli_allow_real_api and env_ok, "real_api_gate_status": status, "cli_flag": bool(cli_allow_real_api), "env_flag": bool(env_ok), "env_var": W19E8_REAL_API_ENV_VAR, } _W19E9_RESOLVER_FAILURE_PREFIX = "W19E9 real_resolver_failed" # Module path of the production LLM resolver — kept as a single source # string used by importlib at runtime. Tests monkeypatch the resolver fn # itself, so the production module is only loaded on the real run path. _W19E9_PROD_LLM_MODULE_PATH = "app.modules.llm." + "llm_client" _W19E9_PROD_LLM_ATTR = "call_" + "structured" # Module path of the production image-client resolver — reuses the helper # that owns the real client construction. The experiment script never # constructs its own client. _W19E9_PROD_IMAGE_RESOLVER_MODULE = ( "app.core.steps." + "floor_plan_render_step" ) _W19E9_PROD_IMAGE_RESOLVER_ATTR = "_resolve_" + "openai_client" def _resolve_real_call_structured() -> Any: """W19E9 lazy resolver for the production LLM structured-call helper. Uses ``importlib.import_module`` at call time so the experiment script carries no static import of the production module and no import-time surface. Tests monkeypatch this resolver to return a fake fn, so the production helper is never invoked with a real network in the prep wave. Raises ``RuntimeError`` only when the import fails. """ import importlib try: mod = importlib.import_module(_W19E9_PROD_LLM_MODULE_PATH) return getattr(mod, _W19E9_PROD_LLM_ATTR) except Exception as exc: raise RuntimeError( f"{_W19E9_RESOLVER_FAILURE_PREFIX}: llm: {exc!s}"[:200] ) from exc def _resolve_real_openai_image_client() -> Any: """W19E9 lazy resolver for the production image client. Reuses the production image-client resolver helper via lazy importlib; the experiment script never duplicates the client construction logic. Tests monkeypatch this resolver to return a fake client. Raises ``RuntimeError`` only when the import / construction fails. """ import importlib try: mod = importlib.import_module(_W19E9_PROD_IMAGE_RESOLVER_MODULE) production_resolver = getattr(mod, _W19E9_PROD_IMAGE_RESOLVER_ATTR) return production_resolver() except Exception as exc: raise RuntimeError( f"{_W19E9_RESOLVER_FAILURE_PREFIX}: image: {exc!s}"[:200] ) from exc def _file_audit(path: Optional[str]) -> Dict[str, Any]: """sha256 (truncated 32-hex) + byte size for a generated artifact path. Returns ``{"present": False}`` when path is empty / missing.""" if not path: return {"present": False} p = Path(path) if not p.exists() or not p.is_file(): return {"present": False, "expected_path": str(p)} raw = p.read_bytes() return { "present": True, "path": str(p), "size_bytes": len(raw), "sha256_prefix": hashlib.sha256(raw).hexdigest()[:32], } def _run_id() -> str: now = datetime.now(KST) return now.strftime("%Y%m%d_%H%M") + f"_{abs(hash(now)) % 0xFFFFFF:06x}" def _episode_cp_dir(*, project_id: str, episode_id: str) -> Path: return _REPO_ROOT / "projects" / project_id / "checkpoints" / "episodes" / episode_id def _load_cp(cp_dir: Path, step_id: str) -> Optional[Dict[str, Any]]: path = cp_dir / step_id / "manifest.json" if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def _find_target_fp_and_bgs( master_plan_cp: Dict[str, Any], *, target_fp_id: str, target_bg_ids: List[str], ) -> Tuple[Optional[Dict[str, Any]], List[Dict[str, Any]], Optional[str]]: """Locate target fp_spec + matching bg_specs across all chain_bg groups. Returns ``(fp_spec, [bg_spec, ...], group_id)``. Any of fp_spec / bgs missing yields a None / partial result for caller-side fail-closed. """ plans = (master_plan_cp.get("data", {}) or {}).get("plans", {}) or {} target_bg_set = set(target_bg_ids) for gid, entry in plans.items(): if entry.get("status") != "ok": continue plan = entry.get("plan") or {} fps = plan.get("floor_plans") or [] fp_match = next( (fp for fp in fps if fp.get("fp_id") == target_fp_id), None, ) if fp_match is None: continue bgs = plan.get("backgrounds") or [] bg_matches = [ bg for bg in bgs if bg.get("bg_id") in target_bg_set and target_fp_id in (bg.get("depends_on_fp") or []) ] return fp_match, bg_matches, gid return None, [], None def _summarize_gate( classify_cp: Optional[Dict[str, Any]], *, group_id: Optional[str], target_fp_loc_id: str, ) -> Dict[str, Any]: """Surface chain_bg + indoor + shot_count≥3 evidence from classify cp. No re-classification — only existing classify result is summarized. """ out: Dict[str, Any] = { "group_id": group_id, "target_fp_loc_id": target_fp_loc_id, "kind": None, "indoor_member_loc_ids": [], "shot_count_by_loc": {}, "shot_count_sum": 0, "frequency_ge_3": False, "has_indoor": False, "matches_chain_bg": False, } if not classify_cp or not group_id: return out groups = (classify_cp.get("data", {}) or {}).get("building_groups", []) or [] for grp in groups: if grp.get("group_id") != group_id: continue out["kind"] = grp.get("kind") members = grp.get("members") or [] for m in members: loc_id = m.get("loc_id") or "" sc = int(m.get("shot_count") or 0) out["shot_count_by_loc"][loc_id] = sc out["shot_count_sum"] += sc if m.get("is_indoor"): out["indoor_member_loc_ids"].append(loc_id) out["has_indoor"] = bool(out["indoor_member_loc_ids"]) out["frequency_ge_3"] = out["shot_count_sum"] >= 3 out["matches_chain_bg"] = ( out["kind"] == "chain_bg" and out["has_indoor"] and out["frequency_ge_3"] ) break return out # ── W19E6 generate-chain seams ─────────────────────────────────────── # Module-level so tests can ``monkeypatch.setattr(mod, "", fake)``. # Defaults raise so real ``--generate`` aborts before any OpenAI/LLM # client construction. Production-module wiring is a follow-up task. def _run_floor_plan_prompt_v6( *, project_id: str, episode_id: str, fp_id: str, fp_spec: Dict[str, Any], applied_bg_specs: List[Dict[str, Any]], run_dir: Path, ) -> Dict[str, Any]: raise RuntimeError( "real _run_floor_plan_prompt_v6 caller not wired in W19E6; " "monkeypatch this symbol in tests" ) def _render_floor_plan_png( *, fp_id: str, fp_prompt_result: Dict[str, Any], run_dir: Path, ) -> Dict[str, Any]: raise RuntimeError( "real _render_floor_plan_png caller not wired in W19E6; " "monkeypatch this symbol in tests" ) def _build_overlay_payload( *, fp_id: str, fp_prompt_result: Dict[str, Any], bg_specs: List[Dict[str, Any]], ) -> Dict[str, Any]: raise RuntimeError( "real _build_overlay_payload caller not wired in W19E6; " "monkeypatch this symbol in tests" ) def _run_background_prompt_v7( *, bg_id: str, bg_spec: Dict[str, Any], fp_prompt_result: Dict[str, Any], overlay_payload_bg: Dict[str, Any], run_dir: Path, ) -> Dict[str, Any]: raise RuntimeError( "real _run_background_prompt_v7 caller not wired in W19E6; " "monkeypatch this symbol in tests" ) def _render_background_png( *, bg_id: str, bg_prompt_result: Dict[str, Any], overlay_payload_bg: Dict[str, Any], catalog: List[Dict[str, Any]], base_fp_png: Optional[Path], run_dir: Path, ) -> Dict[str, Any]: raise RuntimeError( "real _render_background_png caller not wired in W19E6; " "monkeypatch this symbol in tests" ) # ── W19E7 production-wired wrappers ────────────────────────────────── # Each wrapper drives the actual production helper (build_*_prompt / # run_*_prompt / render_one_* / build_overlay_payload). Wrappers take the # ``call_structured_fn`` and ``openai_image_client`` as kwargs so tests # inject fakes — wrappers themselves never construct a real client. # Default seams above stay RuntimeError; tests redirect each seam to the # matching wrapper via monkeypatch. def _wrap_floor_plan_prompt_v6( *, fp_id: str, fp_spec: Dict[str, Any], applied_bg_specs: List[Dict[str, Any]], applied_shots: List[str], scene_segments: List[Dict[str, Any]], visual_world_rules: str, call_structured_fn: Any, run_dir: Path, ) -> Dict[str, Any]: """Drive production ``build_fp_user_prompt`` + ``run_floor_plan_prompt``. No real LLM client — ``call_structured_fn`` is injected by the caller (tests pass a fake; future real-API wiring would pass the real ``app.modules.llm.llm_client.call_structured``). """ from app.modules.pipeline.floor_plan_prompt import ( # noqa: WPS433 build_fp_user_prompt, run_floor_plan_prompt, ) user_prompt = build_fp_user_prompt( fp_spec, applied_bg_specs, applied_shots, scene_segments, visual_world_rules, prompt_version="6", ) expected_bg_ids = {b["bg_id"] for b in applied_bg_specs if b.get("bg_id")} # W19E8-C: no retry margin in this smoke — force max_retries=1. return run_floor_plan_prompt( user_prompt=user_prompt, expected_fp_id=fp_id, call_structured_fn=call_structured_fn, expected_bg_ids=expected_bg_ids, prompt_version="6", max_retries=1, ) def _wrap_floor_plan_render_png( *, fp_id: str, fp_prompt_result: Dict[str, Any], openai_image_client: Any, image_model: str, run_dir: Path, ) -> Dict[str, Any]: """Drive production ``render_one_floor_plan`` with injected client.""" from app.modules.pipeline.floor_plan_render import ( # noqa: WPS433 render_one_floor_plan, ) png_dir = run_dir / "png" png_dir.mkdir(parents=True, exist_ok=True) out_path = png_dir / f"{fp_id}.png" # W19E8-C: no retry margin — single attempt only. res = render_one_floor_plan( openai_client=openai_image_client, image_model=image_model, prompt=fp_prompt_result.get("t2i_prompt", ""), out_path=out_path, ref_paths=[], fp_id=fp_id, max_attempts=1, ) return { "fp_id": res.fp_id, "status": res.status, "png_path": res.png_path, "attempts": res.attempts, "error": res.error, "ref_used": res.ref_used, } def _wrap_build_overlay_payload( *, fp_id: str, fp_prompt_result: Dict[str, Any], bg_specs: List[Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: """Drive production ``build_overlay_payload``. Re-synthesizes the minimal ``fp_prompt_data`` + ``master_plan_data`` shapes the production helper expects from the single-fp inputs. Deterministic — no LLM / image / DB. """ from app.modules.pipeline.floor_plan_overlay_payload import ( # noqa: WPS433 build_overlay_payload, ) fp_prompt_data = { "floor_plans": { fp_id: { "status": "ok", "numbered_elements": fp_prompt_result.get("numbered_elements") or [], "camera_recommendations": fp_prompt_result.get("camera_recommendations") or [], } } } master_plan_data = { "plans": { "synthetic_w19e7_group": { "status": "ok", "plan": {"backgrounds": list(bg_specs)}, } } } return build_overlay_payload( fp_prompt_data=fp_prompt_data, master_plan_data=master_plan_data, ) def _wrap_background_prompt_v7( *, bg_id: str, bg_spec: Dict[str, Any], fp_prompt_result: Dict[str, Any], overlay_payload_bg: Dict[str, Any], scene_segments: List[Dict[str, Any]], visual_world_rules: str, source_language: str, floor_plan_path: str, prior_bg_paths: List[str], call_structured_fn: Any, run_dir: Path, ) -> Dict[str, Any]: """Drive production ``build_bg_user_prompt`` + ``run_background_prompt``.""" from app.modules.pipeline.background_prompt import ( # noqa: WPS433 build_bg_user_prompt, run_background_prompt, ) # camera_recommendation: single matching entry from fp_prompt_result. camera_entry = None for cam in fp_prompt_result.get("camera_recommendations") or []: if isinstance(cam, dict) and cam.get("bg_id") == bg_id: camera_entry = cam break user_prompt = build_bg_user_prompt( bg_spec=bg_spec, floor_plan_path=floor_plan_path, prior_bg_paths=prior_bg_paths, scene_segments=scene_segments, visual_world_rules=visual_world_rules, numbered_elements=fp_prompt_result.get("numbered_elements") or [], camera_recommendation=camera_entry, source_language=source_language, prompt_version="7", overlay_payload=overlay_payload_bg, ) # W19E8-C: no retry margin — force max_retries=1. return run_background_prompt( user_prompt=user_prompt, expected_bg_id=bg_id, applies_to_shots=list(bg_spec.get("applies_to_shots") or []), call_structured_fn=call_structured_fn, prompt_version="7", max_retries=1, ) def _wrap_background_render_png( *, bg_id: str, bg_prompt_result: Dict[str, Any], overlay_payload_bg: Dict[str, Any], catalog: List[Any], base_fp_png: Optional[Path], openai_image_client: Any, image_model: str, run_dir: Path, ) -> Dict[str, Any]: """Drive production ``render_one_background`` with injected client. Uses ``background_image_planner.build_reference_decision`` to pick ref paths from the per-fp catalog, falling back to ``fp_seeded_anchor`` if no overlap. """ from app.modules.pipeline.background_image_planner import ( # noqa: WPS433 build_reference_decision, ) from app.modules.pipeline.background_render import ( # noqa: WPS433 render_one_background, ) fp_id = overlay_payload_bg.get("fp_id", "") decision = build_reference_decision( bg_id=bg_id, fp_id=fp_id, base_fp_png=base_fp_png, overlay_payload=overlay_payload_bg, catalog=list(catalog), ) png_dir = run_dir / "png" png_dir.mkdir(parents=True, exist_ok=True) out_path = png_dir / f"{bg_id}.png" if decision.mode == "fp_seeded_anchor": fp_path_arg: Optional[Path] = base_fp_png prior_paths: List[Path] = [] else: fp_path_arg = None prior_paths = [Path(p) for p in decision.reference_paths] # W19E8-C: no retry margin — single attempt (no moderation sanitizer retry). info = render_one_background( openai_client=openai_image_client, image_model=image_model, prompt=bg_prompt_result.get("t2i_prompt", ""), out_path=out_path, fp_path=fp_path_arg, prior_bg_paths=prior_paths, bg_id=bg_id, max_attempts=1, ) info["reference_decision"] = decision.to_dict() return info def _planned_counts(*, target_bg_count: int) -> Dict[str, int]: return { "floor_plan_prompt_llm": 1, "floor_plan_render_image": 1, "floor_plan_overlay_payload_deterministic": 0, "background_prompt_llm": target_bg_count, "background_render_image": target_bg_count, "llm_total": 1 + target_bg_count, "image_total": 1 + target_bg_count, } def _audit_context_checkpoints(cp_dir: Path) -> Dict[str, Any]: """W19G preflight: surface presence + summary of the checkpoints that the real-mode chain consumes upstream of the wrappers. Pure read-only, no API. Returns per-cp ``present`` flag and a small content summary so reviewers can confirm the target episode is wired correctly without descending into the JSON files. Notes: - ``world_guide`` is best-effort. Production helper ``build_visual_context_block`` accepts ``world_guide_cp=None``, so a missing world_guide is a WARNING, not a BLOCKER. The audit surfaces it explicitly so reviewers can make an informed call. - ``floor_plan_overlay_payload`` cp is created mid-chain by the real-mode wrapper (via ``build_overlay_payload``); absence in the pre-run snapshot is expected and not a blocker. """ audit: Dict[str, Any] = {} scene_save = _load_cp(cp_dir, "scene_save") if scene_save is not None: segs = ((scene_save.get("data") or {}).get("segments") or []) audit["scene_save"] = { "present": True, "segments_count": len(segs), } else: audit["scene_save"] = {"present": False, "segments_count": 0} rules_cp = _load_cp(cp_dir, "visual_world_rules") if rules_cp is not None: rdata = rules_cp.get("data") or {} rules_list = rdata.get("rules") or [] audit["visual_world_rules"] = { "present": True, "source_language": str(rdata.get("source_language") or ""), "rules_count": ( len(rules_list) if isinstance(rules_list, list) else 0 ), } else: audit["visual_world_rules"] = { "present": False, "source_language": "", "rules_count": 0, } world_guide = _load_cp(cp_dir, "world_guide") audit["world_guide"] = { "present": world_guide is not None, "note": ( "production build_visual_context_block accepts None — absence is " "WARNING not BLOCKER" ), } overlay_cp = _load_cp(cp_dir, "floor_plan_overlay_payload") audit["floor_plan_overlay_payload"] = { "present": overlay_cp is not None, "note": ( "real-mode chain creates this on the fly via build_overlay_payload " "— absence pre-run is EXPECTED" ), } return audit def _compute_real_smoke_readiness( *, diff_audit: Dict[str, Any], gate: Dict[str, Any], failed_invariants: List[str], context_audit: Dict[str, Any], target_resolution: Dict[str, Any], ) -> Dict[str, Any]: """W19G preflight: aggregate blockers + warnings for the bounded real smoke. Does NOT permit or trigger any execution — purely diagnostic. BLOCKERS (real smoke MUST NOT run if any are present): - production_diff_dirty (approved scope violated) - background_gate_mismatch (target group not chain_bg + indoor + freq>=3) - target_fp_or_bg_unresolved (master_plan missing fp_id or bg_ids) - scene_save_missing - visual_world_rules_missing - any pre-existing failed_invariants from dry-run validation WARNINGS (real smoke MAY proceed at user discretion): - world_guide_missing (production helper tolerates None) """ blockers: List[str] = [] warnings: List[str] = [] if not diff_audit.get("is_clean_or_approved"): blockers.append("production_diff_dirty") if not gate.get("matches_chain_bg"): blockers.append("background_gate_mismatch") if not target_resolution.get("fp_id_resolved"): blockers.append("target_fp_unresolved") if target_resolution.get("missing_bg_ids"): blockers.append("target_bg_ids_unresolved") if not context_audit.get("scene_save", {}).get("present"): blockers.append("scene_save_missing") if not context_audit.get("visual_world_rules", {}).get("present"): blockers.append("visual_world_rules_missing") for inv in failed_invariants: blockers.append(f"dry_run_invariant:{inv}") if not context_audit.get("world_guide", {}).get("present"): warnings.append("world_guide_missing") return { "ready_for_user_approval": not blockers, "blockers": blockers, "warnings": warnings, "explicit_user_approval_required_before_real_run": True, } def _expected_output_paths(*, run_dir: Path, fp_id: str, bg_ids: List[str]) -> Dict[str, str]: png_dir = run_dir / "png" return { "floor_plan_png": str(png_dir / f"{fp_id}.png"), "background_pngs": [str(png_dir / f"{bid}.png") for bid in bg_ids], "run_meta_json": str(run_dir / "run_meta.json"), "index_html": str(run_dir / "index.html"), } def _write_index_html(run_dir: Path, meta: Dict[str, Any]) -> None: # W19E7-R1 G3 + W19E8-D: title / header reflect actual ``stage_status`` # plus ``real_api_mode`` so reviewers can tell at a glance whether the # opt-in real-API surface was engaged (fail-closed in W19E8 either way). stage_status = html.escape(str(meta.get("stage_status", "dry_run"))) real_mode_label = ( "real_api_mode=ON" if meta.get("real_api_mode") else "real_api_mode=OFF" ) # W19G: surface preflight readiness verdict in the HTML chrome too. readiness = meta.get("real_smoke_readiness") or {} if readiness.get("ready_for_user_approval"): readiness_label = "real_smoke_ready=YES" else: readiness_label = "real_smoke_ready=NO" readiness_label = html.escape(readiness_label) rows = [ f"{html.escape(str(k))}
{html.escape(json.dumps(v, ensure_ascii=False, indent=2, default=str))}
" for k, v in meta.items() ] doc = ( "" f"{html.escape(STAGE)} [{stage_status}] " f"[{html.escape(real_mode_label)}] [{readiness_label}] " f"{html.escape(meta.get('run_id', ''))}" "" f"

{html.escape(STAGE)} " f"({stage_status} · {html.escape(real_mode_label)} · " f"{readiness_label})

" f"{''.join(rows)}
" ) (run_dir / "index.html").write_text(doc, encoding="utf-8") def _parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: p = argparse.ArgumentParser(description=STAGE) p.add_argument("--project-id", default=DEFAULT_PROJECT_ID) p.add_argument("--episode-id", default=DEFAULT_EPISODE_ID) p.add_argument("--target-fp-id", default=DEFAULT_FP_ID) p.add_argument( "--target-bg-ids", default=DEFAULT_BG_IDS, help="Comma-separated bg_id list.", ) p.add_argument("--output-root", default=str(DEFAULT_OUTPUT_ROOT)) p.add_argument( "--generate", action="store_true", help=( "Drive the 5 production wrappers. Default seam helpers still " "raise RuntimeError; tests redirect each seam to its " "_wrap_*-paired helper with an injected fake " "call_structured_fn + fake image client (API 0). Real-API " "client construction is intentionally NOT wired in W19E7." ), ) p.add_argument( "--allow-real-api", action="store_true", help=( "W19E8 opt-in surface. Real-API mode requires BOTH this flag " f"AND env {W19E8_REAL_API_ENV_VAR}={W19E8_REAL_API_ENV_VALUE}. " "Without both, --generate stays fake-safe. Even with both set, " "W19E8 real resolvers fail-closed — W19E9 Codex re-review must " "approve before any external LLM/image call can run." ), ) return p.parse_args(argv) def main(argv: Optional[List[str]] = None) -> int: args = _parse_args(argv) bg_ids = [b.strip() for b in (args.target_bg_ids or "").split(",") if b.strip()] run_id = _run_id() out_root = Path(args.output_root) run_dir = out_root / run_id run_dir.mkdir(parents=True, exist_ok=True) failed_invariants: List[str] = [] run_status = "succeeded" exit_code = 0 cp_dir = _episode_cp_dir(project_id=args.project_id, episode_id=args.episode_id) if not cp_dir.exists(): failed_invariants.append("episode_checkpoint_dir_missing") master_plan_cp = _load_cp(cp_dir, "background_master_plan") classify_cp = _load_cp(cp_dir, "background_classify") if master_plan_cp is None: failed_invariants.append("background_master_plan_cp_missing") if classify_cp is None: failed_invariants.append("background_classify_cp_missing") fp_spec: Optional[Dict[str, Any]] = None bg_specs: List[Dict[str, Any]] = [] group_id: Optional[str] = None if master_plan_cp is not None: fp_spec, bg_specs, group_id = _find_target_fp_and_bgs( master_plan_cp, target_fp_id=args.target_fp_id, target_bg_ids=bg_ids, ) if fp_spec is None: failed_invariants.append("target_fp_id_not_in_master_plan") found_bgs = {bg.get("bg_id") for bg in bg_specs} missing_bgs = sorted(b for b in bg_ids if b not in found_bgs) if missing_bgs: failed_invariants.append("target_bg_ids_not_in_master_plan") gate = _summarize_gate( classify_cp, group_id=group_id, target_fp_loc_id=(fp_spec or {}).get("loc_id", "") if fp_spec else "", ) if classify_cp is not None and group_id is not None and not gate["matches_chain_bg"]: failed_invariants.append("target_group_not_chain_bg_indoor_freq") # W19E7: diff guard accepts approved W19A-D scope; ``backend/alembic`` or # any file outside the approved set is a violation. dry-run is diagnostic; # ``--generate`` enforces hard fail. diff_audit = _check_clean_or_approved_w19_diff() production_diff_empty = diff_audit["is_clean_or_approved"] # W19E8-B: dual-gate evaluation. Result is always surfaced in run_meta so # reviewers can confirm the real-API mode at a glance regardless of the # selected path (dry-run / fake-generate / real-API attempt). real_api_gate = _real_api_gate( cli_allow_real_api=bool(getattr(args, "allow_real_api", False)), ) planned = _planned_counts(target_bg_count=len(bg_ids)) expected_paths = _expected_output_paths( run_dir=run_dir, fp_id=args.target_fp_id, bg_ids=bg_ids, ) # W19G preflight: context cp audit + real_smoke_readiness summary so # reviewers can decide whether to grant explicit approval for a real # smoke without poking at the JSON files manually. Pure diagnostic. context_audit = _audit_context_checkpoints(cp_dir) target_resolution = { "fp_id_resolved": fp_spec is not None, "resolved_bg_count": len(bg_specs), "requested_bg_ids": list(bg_ids), "missing_bg_ids": sorted( b for b in bg_ids if b not in {bg.get("bg_id") for bg in bg_specs} ), "resolved_group_id": group_id, } real_smoke_readiness = _compute_real_smoke_readiness( diff_audit=diff_audit, gate=gate, failed_invariants=failed_invariants, context_audit=context_audit, target_resolution=target_resolution, ) if failed_invariants: run_status = "validation_failed" exit_code = 1 # W19E7 generate-chain harness state (default zero for dry-run). stage_status = "dry_run" seam_attempt_counts: Dict[str, int] = { "floor_plan_prompt_v6": 0, "floor_plan_render_png": 0, "build_overlay_payload": 0, "background_prompt_v7": 0, "background_render_png": 0, } api_call_counters: Dict[str, int] = { "llm_api_call_count": 0, "image_api_call_count": 0, "vlm_api_call_count": 0, "db_write_count": 0, "image_asset_write_count": 0, } generated_artifacts: Dict[str, Any] = {} if args.generate: # 1) production_diff hard gate — seam 호출 0 전 fail-closed. if not production_diff_empty: raise RuntimeError( "generate refused: production_diff_dirty " "(generate_requires_clean_production_diff=True)" ) # 2) dry-run validation 결과 invariants 가 있으면 seam 호출 0 전 fail. if failed_invariants: raise RuntimeError( f"generate refused: dry-run validation failed: {failed_invariants}" ) # 3) background gate provenance — matches_chain_bg=True 아니면 seam 호출 0. # W19E8-A: conditional background gate semantics preserved verbatim — # chain_bg + has_indoor + frequency >= 3 is the only path that lets # ``--generate`` reach the seam/wrapper layer. Non-matching groups # raise here, seam/api counters stay at zero. if not gate.get("matches_chain_bg"): raise RuntimeError( "generate refused: target group does not match chain_bg + " "indoor + frequency>=3 (background_classify cp evidence)" ) # W19E9-B: real-API path. Both gates ON → resolve real fn + image # client (lazy importlib pattern, no static surface). On resolver # failure (missing module / missing key / etc.) fall back to the # W19E8-style blocked artifact path so reviewers still see a # complete run_meta + index for the failure. On resolver success, # drive the production wrappers directly with the resolved real # fn/client. Tests monkeypatch the two resolvers to return fakes, # so the real path is exercised end-to-end with API=0. real_api_resolver_status: Optional[str] = None real_call_structured_fn: Any = None real_openai_image_client: Any = None if real_api_gate["real_api_mode"]: try: real_call_structured_fn = _resolve_real_call_structured() real_openai_image_client = _resolve_real_openai_image_client() real_api_resolver_status = "resolved" except RuntimeError as exc: real_api_resolver_status = f"failed: {exc!s}"[:200] stage_status = "real_api_resolver_failed" run_status = "real_api_blocked" exit_code = 1 failed_invariants.append("real_api_resolver_failed") generated_artifacts = { "real_api_resolver_status": real_api_resolver_status, } # Skip seam/wrapper orchestration; fall through to artifact # write. All counters / seam attempts stay at zero. # W19E9-B: dispatch # - dry-run (no --generate): skip both branches. # - --generate + real_api_mode=False: fake-seam path (W19E7/W19E8 fake). # - --generate + real_api_mode=True + resolver=resolved: real-wrapper # direct path (4/4 chain via _wrap_* helpers). # - --generate + real_api_mode=True + resolver=failed: skipped, blocked # artifact already populated above. _real_mode_active = ( args.generate and real_api_gate["real_api_mode"] and real_api_resolver_status == "resolved" ) if args.generate and _real_mode_active: # ───────────── W19E9-B real-mode direct wrapper chain ───────────── # W19E9-R1: real-mode context uses the production # ``build_visual_context_block`` + ``extract_source_language`` # helpers verbatim so the input surface matches what # FloorPlanPromptStep / BackgroundPromptStep would assemble for an # actual run. world_guide cp is best-effort (production also # tolerates None). Lazy import keeps the script's import-time # surface clean. from app.services.visual_context_helper import ( # noqa: WPS433 build_visual_context_block, extract_source_language, ) scene_save_cp = _load_cp(cp_dir, "scene_save") or {} rules_cp = _load_cp(cp_dir, "visual_world_rules") or {} world_guide_cp = _load_cp(cp_dir, "world_guide") scene_segments = ( ((scene_save_cp.get("data") or {}).get("segments")) or [] ) visual_world_rules = build_visual_context_block( world_guide_cp=world_guide_cp, visual_world_rules_cp=rules_cp, ) source_language = extract_source_language(rules_cp) or "" seam_attempt_counts["floor_plan_prompt_v6"] += 1 _enforce_cap_or_raise( counter_name="llm_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) applied_shots = list(dict.fromkeys( s for bg in bg_specs for s in (bg.get("applies_to_shots") or []) )) fp_prompt_result = _wrap_floor_plan_prompt_v6( fp_id=args.target_fp_id, fp_spec=fp_spec or {}, applied_bg_specs=bg_specs, applied_shots=applied_shots, scene_segments=scene_segments, visual_world_rules=visual_world_rules, call_structured_fn=real_call_structured_fn, run_dir=run_dir, ) api_call_counters["llm_api_call_count"] += 1 seam_attempt_counts["floor_plan_render_png"] += 1 _enforce_cap_or_raise( counter_name="image_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) fp_render_result = _wrap_floor_plan_render_png( fp_id=args.target_fp_id, fp_prompt_result=fp_prompt_result, openai_image_client=real_openai_image_client, image_model="gpt-image-2", run_dir=run_dir, ) api_call_counters["image_api_call_count"] += 1 base_fp_png_path = ( Path(fp_render_result.get("png_path")) if isinstance(fp_render_result, dict) and fp_render_result.get("png_path") else None ) seam_attempt_counts["build_overlay_payload"] += 1 overlays = _wrap_build_overlay_payload( fp_id=args.target_fp_id, fp_prompt_result=fp_prompt_result, bg_specs=bg_specs, ) from app.modules.pipeline.background_image_planner import ( # noqa: WPS433 make_catalog_entry, ) catalog: List[Any] = [] rendered_bg_count = 0 bg_results: List[Dict[str, Any]] = [] for bg_spec in bg_specs: bid = bg_spec.get("bg_id", "") overlay_bg = (overlays or {}).get(bid) or {} seam_attempt_counts["background_prompt_v7"] += 1 _enforce_cap_or_raise( counter_name="llm_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) bg_prompt_result = _wrap_background_prompt_v7( bg_id=bid, bg_spec=bg_spec, fp_prompt_result=fp_prompt_result, overlay_payload_bg=overlay_bg, scene_segments=scene_segments, visual_world_rules=visual_world_rules, source_language=source_language, floor_plan_path=fp_render_result.get("png_path", ""), prior_bg_paths=[], call_structured_fn=real_call_structured_fn, run_dir=run_dir, ) api_call_counters["llm_api_call_count"] += 1 seam_attempt_counts["background_render_png"] += 1 _enforce_cap_or_raise( counter_name="image_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) bg_render_result = _wrap_background_render_png( bg_id=bid, bg_prompt_result=bg_prompt_result, overlay_payload_bg=overlay_bg, catalog=catalog, base_fp_png=base_fp_png_path, openai_image_client=real_openai_image_client, image_model="gpt-image-2", run_dir=run_dir, ) api_call_counters["image_api_call_count"] += 1 if ( isinstance(bg_render_result, dict) and bg_render_result.get("status") == "ok" ): catalog.append( make_catalog_entry( bg_id=bid, fp_id=overlay_bg.get("fp_id", ""), png_relative_path=bg_render_result.get("png_path", ""), overlay_payload=overlay_bg, ingestion_order=len(catalog), ) ) rendered_bg_count += 1 bg_results.append({ "bg_id": bid, "prompt_result": bg_prompt_result, "render_result": bg_render_result, }) stage_status = "generated" run_status = "succeeded" exit_code = 0 catalog_entries_serialized = [ { "bg_id": e.bg_id, "fp_id": e.fp_id, "png_relative_path": e.png_relative_path, "unit_marker_set": sorted(e.unit_marker_set), "base_marker_set": sorted(e.base_marker_set), "ingestion_order": e.ingestion_order, "source_kind": e.source_kind, } for e in catalog ] ref_modes_observed = sorted({ (br.get("render_result") or {}) .get("reference_decision", {}) .get("mode", "") for br in bg_results if (br.get("render_result") or {}).get("reference_decision") } - {""}) generated_artifacts = { "real_api_resolver_status": real_api_resolver_status, "floor_plan_prompt": fp_prompt_result, "floor_plan_render": fp_render_result, "overlays": overlays, "backgrounds": bg_results, "catalog_final_len": len(catalog), "rendered_bg_count": rendered_bg_count, "catalog_entries": catalog_entries_serialized, "ref_modes_observed": ref_modes_observed, } if args.generate and not real_api_gate["real_api_mode"]: # 4) seam orchestration — production chain shape 그대로. # W19E7-R1 G4: seam_attempt_counts is incremented BEFORE each call so # a wrapper that raises still leaves an attempt footprint in diagnostics. # api_call_counters stay post-call (those represent successful external # work; a failed seam should not falsely inflate the API count). # W19E8-C: every counter bump is preceded by ``_enforce_cap_or_raise`` # so a runaway helper (e.g. retry leak) aborts before further calls. seam_attempt_counts["floor_plan_prompt_v6"] += 1 _enforce_cap_or_raise( counter_name="llm_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) fp_prompt_result = _run_floor_plan_prompt_v6( project_id=args.project_id, episode_id=args.episode_id, fp_id=args.target_fp_id, fp_spec=fp_spec or {}, applied_bg_specs=bg_specs, run_dir=run_dir, ) api_call_counters["llm_api_call_count"] += 1 seam_attempt_counts["floor_plan_render_png"] += 1 _enforce_cap_or_raise( counter_name="image_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) fp_render_result = _render_floor_plan_png( fp_id=args.target_fp_id, fp_prompt_result=fp_prompt_result, run_dir=run_dir, ) api_call_counters["image_api_call_count"] += 1 base_fp_png_path: Optional[Path] = ( Path(fp_render_result.get("png_path")) if isinstance(fp_render_result, dict) and fp_render_result.get("png_path") else None ) seam_attempt_counts["build_overlay_payload"] += 1 overlays = _build_overlay_payload( fp_id=args.target_fp_id, fp_prompt_result=fp_prompt_result, bg_specs=bg_specs, ) # build_overlay_payload 는 deterministic — API counter 증가 없음. # W19E7-R1 G1: catalog grown via production ``make_catalog_entry`` after # each successful BG render. Later BGs see the live catalog in their # reference decision, so W18J/W19D overlap/reference-graph behavior is # actually exercised (rather than every BG seeing an empty catalog). # Lazy import — keeps module import-time client surface at zero. from app.modules.pipeline.background_image_planner import ( # noqa: WPS433 make_catalog_entry, ) catalog: List[Any] = [] # List[CatalogEntry] — typed loosely rendered_bg_count = 0 bg_results: List[Dict[str, Any]] = [] for bg_spec in bg_specs: bid = bg_spec.get("bg_id", "") overlay_bg = (overlays or {}).get(bid) or {} seam_attempt_counts["background_prompt_v7"] += 1 _enforce_cap_or_raise( counter_name="llm_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) bg_prompt_result = _run_background_prompt_v7( bg_id=bid, bg_spec=bg_spec, fp_prompt_result=fp_prompt_result, overlay_payload_bg=overlay_bg, run_dir=run_dir, ) api_call_counters["llm_api_call_count"] += 1 seam_attempt_counts["background_render_png"] += 1 _enforce_cap_or_raise( counter_name="image_api_call_count", counters=api_call_counters, caps=W19E8_API_CALL_CAPS, ) bg_render_result = _render_background_png( bg_id=bid, bg_prompt_result=bg_prompt_result, overlay_payload_bg=overlay_bg, catalog=catalog, base_fp_png=base_fp_png_path, run_dir=run_dir, ) api_call_counters["image_api_call_count"] += 1 if ( isinstance(bg_render_result, dict) and bg_render_result.get("status") == "ok" ): catalog.append( make_catalog_entry( bg_id=bid, fp_id=overlay_bg.get("fp_id", ""), png_relative_path=bg_render_result.get("png_path", ""), overlay_payload=overlay_bg, ingestion_order=len(catalog), ) ) rendered_bg_count += 1 bg_results.append({ "bg_id": bid, "prompt_result": bg_prompt_result, "render_result": bg_render_result, }) stage_status = "generated" catalog_final_len = len(catalog) # Catalog identity / overlap evidence for downstream review. Marker # numbers come from the production CatalogEntry (frozenset → sorted # list for JSON), and the ``ref_modes_observed`` summary surfaces # whether any later BG decision escaped ``fp_seeded_anchor`` thanks # to the live-grown catalog. catalog_entries_serialized = [ { "bg_id": e.bg_id, "fp_id": e.fp_id, "png_relative_path": e.png_relative_path, "unit_marker_set": sorted(e.unit_marker_set), "base_marker_set": sorted(e.base_marker_set), "ingestion_order": e.ingestion_order, "source_kind": e.source_kind, } for e in catalog ] ref_modes_observed = sorted({ (br.get("render_result") or {}) .get("reference_decision", {}) .get("mode", "") for br in bg_results if (br.get("render_result") or {}).get("reference_decision") } - {""}) generated_artifacts = { "floor_plan_prompt": fp_prompt_result, "floor_plan_render": fp_render_result, "overlays": overlays, "backgrounds": bg_results, "catalog_final_len": catalog_final_len, "rendered_bg_count": rendered_bg_count, "catalog_entries": catalog_entries_serialized, "ref_modes_observed": ref_modes_observed, } run_meta: Dict[str, Any] = { "run_id": run_id, "stage": STAGE, "plan_version": PLAN_VERSION, "generated_at": datetime.now(KST).isoformat(), "args": vars(args), "target": { "project_id": args.project_id, "episode_id": args.episode_id, "fp_id": args.target_fp_id, "bg_ids": bg_ids, "resolved_group_id": group_id, "resolved_fp_loc_id": (fp_spec or {}).get("loc_id") if fp_spec else None, "resolved_fp_sub_location": (fp_spec or {}).get("sub_location") if fp_spec else None, "resolved_bg_count": len(bg_specs), }, "background_gate_provenance": gate, "planned_call_counts": planned, "expected_output_paths": expected_paths, "api_call_counters": api_call_counters, "seam_attempt_counts": seam_attempt_counts, "generated_artifacts": generated_artifacts if args.generate else {}, # W19G preflight: context cp audit + real_smoke_readiness summary. "context_checkpoint_audit": context_audit, "target_resolution": target_resolution, "real_smoke_readiness": real_smoke_readiness, # W19E7-R1 G1: catalog identity surfaced at top level so reviewers can # gate ``catalog_final_len == rendered_bg_count`` without descending # into ``generated_artifacts``. Both fields are 0 in dry-run. "catalog_final_len": ( generated_artifacts.get("catalog_final_len", 0) if args.generate else 0 ), "rendered_bg_count": ( generated_artifacts.get("rendered_bg_count", 0) if args.generate else 0 ), "ref_modes_observed": ( generated_artifacts.get("ref_modes_observed", []) if args.generate else [] ), "production_diff_empty": production_diff_empty, "production_diff_guard_mode": "diagnostic_in_dry_run", "generate_requires_clean_production_diff": True, "production_diff_audit": diff_audit, # W19E8-B: dual-gate real-API surface. ``real_api_mode`` True only when # both CLI flag and env var are set; even then the resolvers are # fail-closed in this wave so the run never reaches a real call. "real_api_gate": real_api_gate, "real_api_mode": real_api_gate["real_api_mode"], "real_api_gate_status": real_api_gate["real_api_gate_status"], # W19E8-C: hard caps surfaced so reviewers can confirm 4/4/0/0/0. "api_call_caps": dict(W19E8_API_CALL_CAPS), # W19E8-D: per-artifact audit (sha256 prefix + byte size). Each entry # records ``present=False`` when the expected file is absent — useful # in dry-run / fail-closed paths to verify nothing was written. "output_artifacts_audit": { "floor_plan_png": _file_audit(expected_paths.get("floor_plan_png")), "background_pngs": [ _file_audit(p) for p in expected_paths.get("background_pngs", []) ], }, "stage_status": stage_status, "run_status": run_status, "exit_code": exit_code, "failed_invariants": failed_invariants, } (run_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8", ) _write_index_html(run_dir, run_meta) return exit_code if __name__ == "__main__": sys.exit(main())