"""background_pipeline_slice experiment (W1, plan bps_w1).

Goal: production background/image/reference pipeline 연결 가능한 minimal dry-run
slice. LLM 은 raw intent 만, deterministic code 가 bg_catalog helper 로
production-shape adapter plan 빌드. production code 0 수정 / DB write 0 / image
call 0. 9 acceptance gate 통과 시 exit 0, 아니면 validation_failed + exit 1.

# READ-ONLY DB CONTRACT (static-guard enforced by test):
#   This script never invokes any SQLAlchemy Session write callable. DB access
#   is restricted to read-only SessionLocal queries. SQLAlchemy after_flush
#   event listener serves as the runtime sentinel.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import uuid
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any, Dict, List, Optional

PLAN_VERSION = "bps_w1"
DEFAULT_MODEL = "gemini-3.5-flash"
DEFAULT_PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19"
DEFAULT_EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c"

_REPO_ROOT = Path(__file__).resolve().parents[2]
_BACKEND_ROOT = _REPO_ROOT / "backend"
_DEFAULT_OUTPUT_ROOT = _REPO_ROOT / "scripts_output" / "background_pipeline_slice_experiment"

KST = timezone(timedelta(hours=9))

# Make backend importable for production helper imports.
if str(_BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(_BACKEND_ROOT))

# Module-level STATE_CLASS_ENUM import for LLM schema enum (Phase 4).
from app.core.bg_state_vocab import STATE_CLASS_ENUM as _STATE_CLASS_ENUM  # type: ignore  # noqa: E402

# Runtime DB-write sentinel counter (incremented by SQLAlchemy after_flush event).
_DB_WRITE_COUNT = 0


def _run_id() -> str:
    return f"{datetime.now(KST).strftime('%Y%m%d_%H%M')}_{uuid.uuid4().hex[:6]}"


def _load_backend_env() -> None:
    env_path = _BACKEND_ROOT / ".env"
    if not env_path.exists():
        return
    for line in env_path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


def _parse_args(argv):
    p = argparse.ArgumentParser(description="background_pipeline_slice experiment (W1)")
    p.add_argument("--project-id", default=DEFAULT_PROJECT_ID)
    p.add_argument("--episode-id", default=DEFAULT_EPISODE_ID)
    p.add_argument("--output-root", default=str(_DEFAULT_OUTPUT_ROOT))
    p.add_argument("--dry-run", action="store_true", help="default; no LLM call")
    p.add_argument("--generate", action="store_true", help="actual LLM call (gemini-3.5-flash)")
    p.add_argument("--model", default=DEFAULT_MODEL)
    p.add_argument("--skip-db", action="store_true", help="skeleton test only; skip DB load")
    p.add_argument(
        "--stop-after",
        choices=[
            "source_bundle", "pipeline_map", "llm_raw_plan",
            "adapter_plan", "payload", "reference_plan", "compatibility_report",
        ],
        default=None,
    )
    p.add_argument("--diag-print-imports", action="store_true",
                   help="print sorted sys.modules keys to stderr (diagnostic)")
    # W4 — derive render plan from a previous W3 success run dir
    p.add_argument(
        "--derive-render-plan-from",
        default=None,
        help="W4 stage: derive background_render_plan from a previous W3 success run dir "
             "(e.g. scripts_output/background_pipeline_slice_experiment/<run_id>). "
             "Skips W1-W3 phases. No LLM, no DB, no image API.",
    )
    # W5 — derive generation map from a previous W4 success run dir
    p.add_argument(
        "--derive-generation-map-from",
        default=None,
        help="W5 stage: derive background_generation_map from a previous W4 success run dir. "
             "Skips W1-W4 phases. No LLM, no DB, no image API.",
    )
    # W6 — derive image payload plan from a previous W5 success run dir
    p.add_argument(
        "--derive-image-payloads-from",
        default=None,
        help="W6 stage: derive background_image_payload_plan from a previous W5 success run dir. "
             "Skips W1-W5 phases. No LLM, no DB, no image API. Mirrors gpt-image-2 request shape only.",
    )
    # W7 — derive plate prompt plan from a previous W6 success run dir
    p.add_argument(
        "--derive-plate-prompts-from",
        default=None,
        help="W7 stage: derive background_plate_prompt_plan from a previous W6 success run dir. "
             "Skips W1-W6 phases. LLM rewrites payload prompts into background plate prompts. "
             "No DB, no image API. Use --generate to actually call gemini.",
    )
    # W8 — derive image generation from a previous W7 success run dir
    p.add_argument(
        "--derive-image-generation-from",
        default=None,
        help="W8 stage: derive actual gpt-image-2 image generation from a previous W7 success run dir. "
             "Skips W1-W7 phases. Use --generate-images to actually call the gpt-image-2 API.",
    )
    p.add_argument(
        "--generate-images", action="store_true",
        help="W8 only: when set, actually invoke gpt-image-2 API and write PNG files into the run dir. "
             "Without this flag, W8 dry-run validates payload shape only; no network, no PNG.",
    )
    p.add_argument(
        "--selected-bg-ids", default=None,
        help="W8 optional: comma-separated subset of bg_ids to generate. "
             "Default = all bg_ids in the W7 plan. Order follows the W7 render_batches.",
    )
    # W8b — opt-in policy switch for how reference_strength=='weak' payloads
    # are actually called against gpt-image-2. Generic flag; no scenario rule.
    p.add_argument(
        "--weak-reference-mode",
        choices=["edit_with_ref", "prompt_only"],
        default="edit_with_ref",
        help="W8 weak-payload policy. 'edit_with_ref' (default): weak variants still call "
             "images.edit with the declared reference PNG (W8 behavior). 'prompt_only': "
             "weak variants skip the reference image entirely and call images.generate "
             "based on plate_prompt_text alone. strict variants always use images.edit; "
             "independent payloads always use images.generate.",
    )
    # W8h — derive image_continuity_subgroups via LLM from a W7 run.
    p.add_argument(
        "--derive-image-continuity-from",
        default=None,
        help="W8h stage: derive image_continuity_subgroups from a prior W7 success run dir. "
             "LLM groups bg_ids that share the same physical subspace so that same-room "
             "state transitions keep their direct parent reference. No DB, no image API.",
    )
    # W9 / W8h-aware — feed an image_continuity_subgroups run into W8.
    p.add_argument(
        "--continuity-from",
        default=None,
        help="W8 optional: path to a prior W8h run dir whose image_continuity_subgroups.json "
             "will override --weak-reference-mode for same-subgroup direct parents. Cross-"
             "subgroup weak payloads still follow --weak-reference-mode.",
    )
    # W11 — derive director_set_brief from a prior W7 success run (read-only).
    # Loads production floor_plan_prompt + floor_plan_render manifest as
    # strong-soft / weak-visual evidence. PNG bytes are NOT read. LLM produces
    # group_set_briefs + per_bg_set_selections + final_plate_prompt_candidates.
    p.add_argument(
        "--derive-director-set-brief-from",
        default=None,
        help="W11 stage: derive director_set_brief plan from a prior W7 success run dir. "
             "Loads production floor_plan_prompt JSON (strong soft evidence) + "
             "floor_plan_render PNG path metadata (weak visual evidence, path only — no bytes). "
             "LLM reconciles per-bg relevant_numbered_elements + camera_axis + final plate "
             "prompt candidate. Use --generate to actually call the LLM. No DB, no image API.",
    )
    return p.parse_args(argv)


def main(argv=None) -> int:
    global _DB_WRITE_COUNT
    args = _parse_args(argv)
    run_id = _run_id()
    out_root = Path(args.output_root)
    run_dir = out_root / run_id
    run_dir.mkdir(parents=True, exist_ok=True)

    # W4 stage — derive render plan from a prior W3 success run. Bypasses W1-W3.
    if args.derive_render_plan_from:
        return _w4_main(args, run_dir, run_id)

    # W5 stage — derive generation map from a prior W4 success run. Bypasses W1-W4.
    if args.derive_generation_map_from:
        return _w5_main(args, run_dir, run_id)

    # W6 stage — derive image payload plan from a prior W5 success run. Bypasses W1-W5.
    if args.derive_image_payloads_from:
        return _w6_main(args, run_dir, run_id)

    # W7 stage — derive plate prompt plan from a prior W6 success run. Bypasses W1-W6.
    if args.derive_plate_prompts_from:
        return _w7_main(args, run_dir, run_id)

    # W8 stage — derive image generation from a prior W7 success run. Bypasses W1-W7.
    if args.derive_image_generation_from:
        return _w8_main(args, run_dir, run_id)

    # W8h stage — derive image_continuity_subgroups via LLM from a prior W7 run.
    if args.derive_image_continuity_from:
        return _w8h_main(args, run_dir, run_id)

    # W11 stage — derive director_set_brief from a prior W7 success run.
    if args.derive_director_set_brief_from:
        return _w11_main(args, run_dir, run_id)

    outputs: List[str] = []
    failed_invariants: List[str] = []
    run_status = "succeeded"
    exit_code = 0
    bundle: Optional[dict] = None
    raw_plan: Optional[dict] = None
    adapter: Optional[dict] = None
    rip: Optional[dict] = None
    ref_dry: Optional[dict] = None
    report: Optional[dict] = None
    pmap_check: Optional[dict] = None

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model": args.model if args.generate else None,
        "args": vars(args),
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed_invariants,
    }

    def _write_run_meta() -> None:
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed_invariants
        run_meta["outputs"] = outputs
        if pmap_check is not None:
            run_meta["production_pipeline_map_check"] = pmap_check
        (run_dir / "run_meta.json").write_text(
            json.dumps(run_meta, ensure_ascii=False, indent=2)
        )

    # ── Phase 2: source_bundle ────────────────────────────────────────
    if not args.skip_db:
        _install_db_write_sentinel()
        bundle = _load_source_bundle(args.project_id, args.episode_id)
        (run_dir / "source_bundle.json").write_text(
            json.dumps(bundle, ensure_ascii=False, indent=2)
        )
        outputs.append("source_bundle.json")
        if bundle["selected_shot_count"] == 0:
            failed_invariants.append("no_selected_shots")
            run_status = "validation_failed"
            exit_code = 1
        if args.stop_after == "source_bundle":
            _write_run_meta()
            _maybe_print_imports(args)
            return exit_code

    # ── Phase 3: production_pipeline_map ─────────────────────────────
    pmap_check = _verify_pipeline_map(run_dir)
    outputs.append("production_pipeline_map.json")
    if not (pmap_check["all_paths_exist"] and pmap_check["all_symbols_found"]):
        failed_invariants.append("production_pipeline_map_stale")
        run_status = "validation_failed"
        exit_code = 1
    if args.stop_after == "pipeline_map":
        _write_run_meta()
        _maybe_print_imports(args)
        return exit_code

    # ── Phase 4: llm_raw_plan ────────────────────────────────────────
    bundle_for_plan = (
        bundle if bundle is not None
        else {"selected_shots": [], "locations": [], "locations_for_llm": [],
              "location_profiles": {}}
    )
    location_profiles_for_validate = bundle_for_plan.get("location_profiles", {}) or {}

    def _validate_generated_plan(plan_obj: dict) -> None:
        """schema + ID membership + semantic-key parent + space_key_hint membership.

        모두 production helper 기반 deterministic check — 어느 하나라도 fail 면 raise
        (retry trigger). regex/semantic 판단 아님.
        """
        import jsonschema  # type: ignore
        jsonschema.validate(plan_obj["plan"], LLM_RAW_PLAN_SCHEMA)
        if not location_profiles_for_validate:
            return
        # W2 — structured loc_id ID membership
        errors = _validate_loc_id_membership(
            plan_obj["plan"], location_profiles_for_validate
        )
        # W3 — space_key_hint allowed membership
        errors.extend(
            _validate_space_key_hint_membership(
                plan_obj["plan"], location_profiles_for_validate
            )
        )
        # W3 BLOCKING — semantic-key parent contract
        errors.extend(
            _validate_semantic_key_parents(
                plan_obj["plan"], location_profiles_for_validate
            )
        )
        if errors:
            head = "; ".join(errors[:5])
            tail = f" (+{len(errors)-5} more)" if len(errors) > 5 else ""
            raise ValueError(f"raw_plan_validation_failed: {head}{tail}")

    if args.generate:
        try:
            raw_plan = _generate_llm_raw_plan(bundle_for_plan, model=args.model)
            _validate_generated_plan(raw_plan)
        except Exception:
            try:
                raw_plan = _generate_llm_raw_plan(bundle_for_plan, model=args.model)
                _validate_generated_plan(raw_plan)
                raw_plan["correction_attempted"] = True
            except Exception as exc2:
                raw_plan = {
                    "status": "generate_failed",
                    "plan": None,
                    "model_used": args.model,
                    "error": str(exc2),
                }
                failed_invariants.append("llm_raw_plan_generate_failed")
                run_status = "validation_failed"
                exit_code = 1
    else:
        raw_plan = _build_placeholder_raw_plan(bundle_for_plan)
    (run_dir / "llm_raw_plan.json").write_text(
        json.dumps(raw_plan, ensure_ascii=False, indent=2)
    )
    outputs.append("llm_raw_plan.json")
    if args.stop_after == "llm_raw_plan":
        _write_run_meta()
        _maybe_print_imports(args)
        return exit_code

    # ── Phase 5: production_adapter_plan ──────────────────────────────
    plan_dict = raw_plan.get("plan") if raw_plan else None
    location_profiles = bundle.get("location_profiles", {}) if bundle else {}
    if plan_dict is None:
        adapter = {
            "schema_version": 1,
            "plan_version": PLAN_VERSION,
            "building_groups": [],
            "floor_plans": [],
            "background_catalog": {},
            "shot_background_map": {},
            "bg_catalog_hash": "",
            "shot_binding_hash": "",
            "gen_order": [],
            "intent_key_to_bg_id": {},
        }
    else:
        try:
            adapter = _build_adapter_plan(plan_dict, location_profiles)
        except Exception as exc:
            adapter = {
                "schema_version": 1,
                "plan_version": PLAN_VERSION,
                "building_groups": [],
                "floor_plans": [],
                "background_catalog": {},
                "shot_background_map": {},
                "bg_catalog_hash": "",
                "shot_binding_hash": "",
                "gen_order": [],
                "intent_key_to_bg_id": {},
                "error": str(exc),
            }
            failed_invariants.append("adapter_plan_build_failed")
            run_status = "validation_failed"
            exit_code = 1
    (run_dir / "production_adapter_plan.json").write_text(
        json.dumps(adapter, ensure_ascii=False, indent=2)
    )
    outputs.append("production_adapter_plan.json")
    if args.stop_after == "adapter_plan":
        _write_run_meta()
        _maybe_print_imports(args)
        return exit_code

    # ── Phase 6: generation_payload_dry_run + reference_input_plan ───
    payload = _build_generation_payload(plan_dict or {}, adapter)
    (run_dir / "generation_payload_dry_run.json").write_text(
        json.dumps(payload, ensure_ascii=False, indent=2)
    )
    outputs.append("generation_payload_dry_run.json")
    if args.stop_after == "payload":
        _write_run_meta()
        _maybe_print_imports(args)
        return exit_code

    rip = _build_reference_input_plan(plan_dict or {}, adapter)
    (run_dir / "reference_input_plan.json").write_text(
        json.dumps(rip, ensure_ascii=False, indent=2)
    )
    outputs.append("reference_input_plan.json")
    if args.stop_after == "reference_plan":
        _write_run_meta()
        _maybe_print_imports(args)
        return exit_code

    # ── Phase 7: pipeline_compatibility_report ────────────────────────
    ref_dry = _ref_contract_dry_run(rip, adapter)
    report = _build_compatibility_report(
        bundle=bundle if bundle is not None else {
            "selected_shots": [], "selected_shot_count": 0, "source_hash": "",
        },
        raw_plan=plan_dict or {},
        adapter=adapter,
        rip=rip,
        ref_dry=ref_dry,
        pmap_check=pmap_check,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_import_seen=_check_image_imports_present(),
    )
    (run_dir / "pipeline_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("pipeline_compatibility_report.json")
    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed_invariants:
            failed_invariants.append(name)
    if failed_invariants:
        run_status = "validation_failed"
        exit_code = 1
        if raw_plan and raw_plan.get("status") == "generated":
            (run_dir / "llm_raw_plan_quarantined.json").write_text(
                json.dumps(raw_plan, ensure_ascii=False, indent=2)
            )
            outputs.append("llm_raw_plan_quarantined.json")
    if args.stop_after == "compatibility_report":
        _write_run_meta()
        _maybe_print_imports(args)
        return exit_code

    # ── Phase 8: HTML diagnostic ──────────────────────────────────────
    _render_html(run_meta, bundle, adapter, rip, report, run_dir)
    outputs.append("index.html")

    _write_run_meta()
    _maybe_print_imports(args)
    return exit_code


def _maybe_print_imports(args) -> None:
    if getattr(args, "diag_print_imports", False):
        sys.stderr.write(",".join(sorted(sys.modules.keys())) + "\n")


# ──────────────────────────────────────────────────────────────────────
# Phase 2 — SourceBundle Loader
# ──────────────────────────────────────────────────────────────────────


def _load_source_bundle(project_id: str, episode_id: str) -> dict:
    """DB read-only. SessionLocal + scope 안에서 모든 데이터 fetch 후 dict 변환.
    READ-ONLY guard: SQLAlchemy after_flush event listener counts writes.
    """
    _load_backend_env()
    from app.core.database import SessionLocal  # type: ignore
    from app.models.catalog import ProjectRegistry  # type: ignore
    from app.models.project import Episode, SceneStill, EntityCanon  # type: ignore
    from app.core.bg_state_vocab import validate_location_space_profile  # type: ignore

    with SessionLocal() as session:
        proj = session.query(ProjectRegistry).filter_by(id=project_id).one_or_none()
        ep = session.query(Episode).filter_by(id=episode_id, project_id=project_id).one_or_none()
        if proj is None or ep is None:
            empty = {
                "project_id": project_id,
                "episode_id": episode_id,
                "project_name": None,
                "episode_title": None,
                "episode_number": None,
                "selected_shot_count": 0,
                "selected_shots": [],
                "locations": [],
                "location_profiles": {},
                "location_profile_errors": [],
                "source_hash": "",
            }
            return empty
        rows = (
            session.query(SceneStill)
            .filter_by(project_id=project_id, episode_id=episode_id, is_selected=True)
            .order_by(SceneStill.scene_index, SceneStill.shot_index, SceneStill.still_index)
            .all()
        )
        shots = []
        for r in rows:
            scene_idx = r.scene_index if r.scene_index is not None else 0
            shot_idx = r.shot_index if r.shot_index is not None else 0
            shot_key = f"S{scene_idx:02d}_Shot{shot_idx}"
            shots.append({
                "row_id": r.id,
                "shot_key": shot_key,
                "project_id": r.project_id,
                "episode_id": r.episode_id,
                "still_index": r.still_index,
                "scene_index": r.scene_index,
                "shot_index": r.shot_index,
                "shot_description": r.shot_description,
                "screenplay_scene_heading": r.screenplay_scene_heading,
                "beat_title": r.beat_title,
                "still_frame_prompt": r.still_frame_prompt,
                "camera_json": r.camera_json,
                "lighting_json": r.lighting_json,
                "visible_entities_json": r.visible_entities_json,
                "t2i_prompt_cinematic": r.t2i_prompt_cinematic,
                "t2i_prompt_closeup": r.t2i_prompt_closeup,
                "t2i_variations_json": r.t2i_variations_json,
                "dependent_scene_id": r.dependent_scene_id,
                "shot_type_1": r.shot_type_1,
                "shot_type_2": r.shot_type_2,
                "scene_type": r.scene_type,
                "scene_summary": r.scene_summary,
                "is_selected": bool(r.is_selected),
            })

        locs = (
            session.query(EntityCanon)
            .filter_by(project_id=project_id, entity_type="location")
            .all()
        )
        location_rows = []
        location_profiles: Dict[str, dict] = {}
        profile_errors = []
        for L in locs:
            row = {
                "id": L.id,
                "short_id": L.short_id,
                "entity_type": L.entity_type,
                "name": L.name,
                "description": L.description,
                "metadata_json": L.metadata_json,
            }
            location_rows.append(row)
            try:
                meta = L.metadata_json
                if isinstance(meta, str):
                    meta = json.loads(meta) if meta else {}
                profile = validate_location_space_profile(meta, short_id=L.short_id)
                location_profiles[L.short_id] = profile
            except Exception as exc:
                profile_errors.append({"short_id": L.short_id, "error": str(exc)})

    # W2 BLOCKING — LLM prompt 용 slim view. UUID `id` 제거, loc_id = short_id.
    # LLM 이 UUID 를 loc_id 로 잘못 고르는 일 차단. raw `locations` 는 진단용 보존.
    locations_for_llm = [
        {
            "loc_id": row["short_id"],
            "name": row["name"],
            "description": row["description"],
            "space_profile": location_profiles[row["short_id"]],
        }
        for row in location_rows
        if row["short_id"] in location_profiles
    ]

    bundle = {
        "project_id": project_id,
        "episode_id": episode_id,
        "project_name": proj.name,
        "episode_title": ep.title,
        "episode_number": ep.episode_number,
        "selected_shot_count": len(shots),
        "selected_shots": shots,
        "locations": location_rows,
        "locations_for_llm": locations_for_llm,
        "location_profiles": location_profiles,
        "location_profile_errors": profile_errors,
    }
    canon = json.dumps(shots, sort_keys=True, ensure_ascii=False).encode("utf-8")
    bundle["source_hash"] = hashlib.sha256(canon).hexdigest()[:16]
    return bundle


# ──────────────────────────────────────────────────────────────────────
# Phase 3 — production_pipeline_map
# ──────────────────────────────────────────────────────────────────────


PRODUCTION_PIPELINE_MAP = {
    "touchpoints": [
        {
            "file": "backend/app/core/bg_catalog.py",
            "symbol": "assign_bg_ids",
            "purpose": "Deterministic L##B## bg_id assignment from LLM raw intents",
            "observed_contract": (
                "assign_bg_ids(prev_catalog: Dict[str, Dict[str, Any]], "
                "new_intents: List[Dict[str, Any]], "
                "location_profiles: Dict[loc_id, profile]) -> Dict[bg_id, entry]. "
                "entry = {bg_id, loc_id, space_key, time_phase, state_class, semantic_key, "
                "depends_on_fp[], depends_on_bg[], applies_to_shots[], sub_location_label, "
                "state_label_raw}. Raises SemanticKeyError/FpLinkMismatchError. Same "
                "semantic_key with differing render-relevant deps -> SemanticKeyError fail-fast."
            ),
        },
        {
            "file": "backend/app/core/bg_catalog.py",
            "symbol": "build_shot_background_map",
            "purpose": "Flatten catalog applies_to_shots into shot_id->bg_id; N:1 invariant",
            "observed_contract": (
                "build_shot_background_map(catalog: Dict[bg_id, entry]) -> Dict[shot_id, bg_id]. "
                "Raises ShotBindingError if same shot maps to >1 bg_id."
            ),
        },
        {
            "file": "backend/app/core/bg_catalog.py",
            "symbol": "compute_bg_catalog_hash",
            "purpose": "SHA256[:16] over render-relevant catalog fields for drift detection",
            "observed_contract": (
                "compute_bg_catalog_hash(catalog: Dict[bg_id, entry]) -> str. "
                "Excludes applies_to_shots/sub_location_label/state_label_raw."
            ),
        },
        {
            "file": "backend/app/core/bg_state_vocab.py",
            "symbol": "BG_ID_RE",
            "purpose": (
                "Production bg_id format regex (^L\\d{2,3}B\\d{2,3}$) - "
                "ID-format validation only, not semantic"
            ),
            "observed_contract": (
                "re.compile(r'^L\\d{2,3}B\\d{2,3}$'). Used by background_render_step "
                "strict marker filter."
            ),
        },
        {
            "file": "backend/app/core/bg_state_vocab.py",
            "symbol": "STATE_CLASS_ENUM",
            "purpose": "11-value state_class vocabulary for background variants",
            "observed_contract": (
                "Frozenset of allowed state_class strings emitted by master_plan LLM "
                "and validated by assign_bg_ids."
            ),
        },
        {
            "file": "backend/app/core/ref_contract_validator.py",
            "symbol": "validate_attached_refs",
            "purpose": "Tier-3 attached-reference identity contract; RefContractError HTTP422 on violation",
            "observed_contract": (
                "validate_attached_refs(rpc: dict, labeled_refs: List[(label, bytes)], "
                "attached_meta: List[(kind, id)], prompt: str, is_close_framing: bool, "
                "*, chain_bg_lookup: Callable, reference_phrase_kinds: List[str]) -> None. "
                "6 sequential checks (length, rpc shape, character_outlook strict, prop strict, "
                "character base strict, background lineage + close-framing skip + readiness, "
                "phantom guard)."
            ),
        },
        {
            "file": "backend/app/core/steps/background_master_plan_step.py",
            "symbol": "BackgroundMasterPlanStep",
            "purpose": (
                "Production producer that emits plans[group_id]={floor_plans, backgrounds, "
                "gen_order} + shot_background_map (mirror target for production_adapter_plan)"
            ),
            "observed_contract": (
                "SCHEMA_VERSION=3, PROMPT_VERSION='5.202605201759'. _d6_post_process calls "
                "bg_catalog.assign_bg_ids. Output checkpoint at "
                "projects/{pid}/checkpoints/episodes/{eid}/background_master_plan/manifest.json."
            ),
        },
        {
            "file": "backend/app/core/steps/background_render_step.py",
            "symbol": "BackgroundRenderStep",
            "purpose": (
                "PNG producer; ImageAsset(asset_type='chain_bg', variant_type=bg_id) "
                "(do NOT call from experiment)"
            ),
            "observed_contract": (
                "Consumes background_master_plan + background_prompt + floor_plan_render. "
                "UPSERT pattern. gpt-image-2 model. Strict BG_ID_RE marker filter."
            ),
        },
        {
            "file": "backend/app/services/scene_checkpoint_loaders.py",
            "symbol": "load_background_chain_bg_map",
            "purpose": (
                "Image-phase consumer that produces "
                "{scene_idx_shot_idx: {bg_id, location_id, label, image_bytes}}"
            ),
            "observed_contract": (
                "load_background_chain_bg_map(project_id, episode_id) -> Dict[shot_key, dict]. "
                "Reads background_render checkpoint; respects settings.background_chain_enabled."
            ),
        },
        {
            "file": "backend/app/services/scene_generation_coordinator.py",
            "symbol": "build_chain_bg_lookup",
            "purpose": "Build callable bg_id->location_id for ref_contract_validator chain_bg_lookup",
            "observed_contract": (
                "build_chain_bg_lookup(background_chain_bg_map) -> Callable[[bg_id], Optional[loc_id]]."
            ),
        },
    ],
}


def _verify_pipeline_map(run_dir: Path) -> dict:
    out_path = run_dir / "production_pipeline_map.json"
    out_path.write_text(json.dumps(PRODUCTION_PIPELINE_MAP, ensure_ascii=False, indent=2))
    all_paths_exist = True
    all_symbols_found = True
    missing = []
    for tp in PRODUCTION_PIPELINE_MAP["touchpoints"]:
        p = _REPO_ROOT / tp["file"]
        if not p.exists():
            all_paths_exist = False
            missing.append({"path": tp["file"], "kind": "path_missing"})
            continue
        if tp["symbol"] not in p.read_text():
            all_symbols_found = False
            missing.append({
                "path": tp["file"],
                "symbol": tp["symbol"],
                "kind": "symbol_missing",
            })
    return {
        "all_paths_exist": all_paths_exist,
        "all_symbols_found": all_symbols_found,
        "missing": missing,
    }


# ──────────────────────────────────────────────────────────────────────
# Phase 4 — LLM Raw Plan
# ──────────────────────────────────────────────────────────────────────


LLM_RAW_PLAN_SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "required": [
        "building_groups", "raw_background_intents", "floor_plan_intents",
        "shot_binding_intents", "prompt_payload_intents", "open_risks",
    ],
    "properties": {
        "building_groups": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["group_id", "member_loc_ids", "anchor_loc_id", "kind", "rationale"],
                "properties": {
                    "group_id": {"type": "string"},
                    "member_loc_ids": {"type": "array", "items": {"type": "string"}},
                    "anchor_loc_id": {"type": "string"},
                    "kind": {"type": "string", "enum": ["chain_bg", "prev_shot_ref", "skip"]},
                    "rationale": {"type": "string"},
                },
            },
        },
        "raw_background_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                # NOTE: applies_to_shots 필드 없음 — shot_binding_intents 가 SOT.
                # code 가 binding 으로부터 derive 후 catalog.applies_to_shots 에 채워넣음.
                "required": [
                    "intent_key", "group_id", "loc_id", "space_key_hint",
                    "time_phase", "state_class", "sub_location_label",
                    "state_label_raw", "fp_intent_key",
                    "parent_intent_ref", "evidence_basis",
                ],
                "properties": {
                    "intent_key": {
                        "type": "string",
                        "description": (
                            "stable key (lowercase a-z0-9_) for cross-reference; NOT bg_id"
                        ),
                    },
                    "group_id": {"type": "string"},
                    "loc_id": {"type": "string"},
                    "space_key_hint": {"type": "string"},
                    "time_phase": {"type": "string"},
                    "state_class": {"type": "string", "enum": sorted(_STATE_CLASS_ENUM)},
                    "sub_location_label": {"type": ["string", "null"]},
                    "state_label_raw": {"type": ["string", "null"]},
                    "fp_intent_key": {"type": ["string", "null"]},
                    "parent_intent_ref": {"type": ["string", "null"]},
                    "evidence_basis": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "additionalProperties": False,
                            "required": ["quote", "source_ref", "confidence"],
                            "properties": {
                                "quote": {"type": "string"},
                                "source_ref": {"type": "string"},
                                "confidence": {
                                    "type": "string",
                                    "enum": ["trusted", "plausible", "weak"],
                                },
                            },
                        },
                    },
                },
            },
        },
        "floor_plan_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["fp_intent_key", "loc_id", "space_key_hint", "rationale"],
                "properties": {
                    "fp_intent_key": {"type": "string"},
                    "loc_id": {"type": "string"},
                    "space_key_hint": {"type": "string"},
                    "rationale": {"type": "string"},
                },
            },
        },
        "shot_binding_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["shot_key", "intent_key"],
                "properties": {
                    "shot_key": {
                        "type": "string",
                        "description": (
                            "composite shot id (e.g. 'S{scene_index}_Shot{shot_index}'). "
                            "SOT for all binding."
                        ),
                    },
                    "intent_key": {"type": "string"},
                },
            },
        },
        "prompt_payload_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["shot_key", "bg_intent_text"],
                "properties": {
                    "shot_key": {"type": "string"},
                    "bg_intent_text": {"type": "string"},
                },
            },
        },
        "open_risks": {
            "type": "array",
            "maxItems": 5,
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["type", "severity", "description", "conservative_fallback"],
                "properties": {
                    "type": {"type": "string"},
                    "severity": {"type": "string", "enum": ["low", "med", "high"]},
                    "description": {"type": "string"},
                    "conservative_fallback": {"type": "string"},
                },
            },
        },
    },
}


def _build_placeholder_raw_plan(bundle: dict) -> dict:
    """dry-run 기본값. 실제 LLM call 없이 schema-valid skeleton 만 emit."""
    plan = {
        "building_groups": [],
        "raw_background_intents": [],
        "floor_plan_intents": [],
        "shot_binding_intents": [],
        "prompt_payload_intents": [],
        "open_risks": [
            {
                "type": "placeholder_dry_run",
                "severity": "low",
                "description": "No LLM call performed; --generate not specified.",
                "conservative_fallback": (
                    "Run with --generate --model gemini-3.5-flash to populate plan."
                ),
            }
        ],
    }
    return {"status": "placeholder_dry_run", "plan": plan, "model_used": None}


def _generate_llm_raw_plan(bundle: dict, *, model: str) -> dict:
    """--generate 시 호출. litellm 직접 + response_format json_object."""
    import litellm  # lazy import — dry-run 에서는 absent

    system_prompt = (
        "You plan production-compatible background generation units for a film scene "
        "pipeline. Emit raw semantic intents only - do NOT assign bg_id, floor_plan id, "
        "or any deterministic identifier; downstream code assigns those via the production "
        "helper. Each background intent must include loc_id, space_key_hint, time_phase, "
        "state_class, sub_location_label, state_label_raw, and evidence_basis quoting the "
        "source bundle. shot-to-intent mapping belongs to shot_binding_intents (one entry "
        "per selected shot_key), not in raw_background_intents. Use intent_key for cross-"
        "reference and parent_intent_ref for depends_on_bg semantics. "
        # W2 BLOCKING — loc_id contract:
        "loc_id MUST be an exact value from the provided locations[].loc_id entries; "
        "do NOT invent loc_id values, do NOT use any database UUID, row_id, name, or other "
        "field as loc_id. space_key_hint MUST be one of the location's space_profile."
        "allowed_space_keys (or 'main' when kind=single_space). "
        # W3 BLOCKING — semantic-key dedup rule:
        "Production merges every intent sharing the same 4-tuple "
        "(loc_id, normalized space_key_hint, time_phase, state_class) into one background id. "
        "sub_location_label and state_label_raw are metadata only and do NOT split background ids. "
        "If two shots differ only by sub-location detail within the same 4-tuple, "
        "bind them to the SAME intent_key and express shot-specific detail in "
        "prompt_payload_intents.bg_intent_text. parent_intent_ref MUST point to an intent "
        "with a DIFFERENT 4-tuple semantic key; pointing to an intent that shares the same "
        "4-tuple creates a self-cycle and will be rejected. "
        "Constraints: do not "
        "introduce sample-specific location labels, named characters, room labels, props, "
        "or episode identifiers as methodology rules; source quotes may contain fixture "
        "data, but your own schema/rules must stay generic. Max 5 open_risks, each with a "
        "conservative_fallback. Output must match the provided JSON schema."
    )
    user_prompt = json.dumps(
        {
            "source_bundle": {
                "selected_shots": bundle.get("selected_shots", []),
                # W2 — UUID 없는 slim view; LLM 이 보는 location SOT.
                "locations": bundle.get("locations_for_llm", []),
            },
            "json_schema": LLM_RAW_PLAN_SCHEMA,
        },
        ensure_ascii=False,
    )
    routed = (
        model
        if model.startswith("gemini/") or not model.lower().startswith("gemini")
        else f"gemini/{model}"
    )
    if not (os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")):
        raise RuntimeError("missing GEMINI_API_KEY / GOOGLE_API_KEY env var")
    resp = litellm.completion(
        model=routed,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        response_format={"type": "json_object"},
    )
    raw_text = resp.choices[0].message.content
    # W2 narrow — Gemini 가 response_format=json_object 임에도 trailing text/extra
    # JSON 을 붙이는 경우 발견. first valid JSON object 만 추출 (raw_decode).
    decoder = json.JSONDecoder()
    stripped = raw_text.lstrip()
    parsed, end = decoder.raw_decode(stripped)
    trailing_ignored = bool(stripped[end:].strip())  # W3 — diagnostic honesty
    return {
        "status": "generated",
        "plan": parsed,
        "model_used": model,
        "raw_text_trailing_ignored": trailing_ignored,
    }


def _compute_intent_semantic_keys(raw_plan: dict, location_profiles: dict) -> Dict[str, str]:
    """W3 — production helper 그대로 사용해 intent_key → semantic_key 매핑.

    같은 semantic_key 인 intent 들은 production 에서 single bg_id 로 dedup 됨.
    LLM 이 이 dedup 규칙을 위반하지 않게 사전 검증 input.
    Returns {intent_key: semantic_key}. invalid intent (loc_id miss 등) 는 제외.
    """
    sys.path.insert(0, str(_BACKEND_ROOT))
    from app.core import bg_catalog  # type: ignore

    out: Dict[str, str] = {}
    for it in raw_plan.get("raw_background_intents", []) or []:
        ikey = it.get("intent_key")
        loc = it.get("loc_id")
        hint = it.get("space_key_hint")
        time_phase = it.get("time_phase")
        state_class = it.get("state_class")
        if not ikey or loc not in location_profiles:
            continue
        try:
            space_key = bg_catalog.normalize_space_key(loc, hint, location_profiles[loc])
            sem_key = bg_catalog.compute_semantic_key(loc, space_key, time_phase, state_class)
        except Exception:
            continue
        out[ikey] = sem_key
    return out


def _validate_semantic_key_parents(raw_plan: dict, location_profiles: dict) -> List[str]:
    """W3 BLOCKING — production `(loc_id, space_key, time_phase, state_class)` dedup
    규칙을 LLM 출력에 사전 적용.

    Two failure modes (둘 다 production assign_bg_ids 도달 전 잡아서 retry 가능):
    - parent_same_semantic_key: parent_intent_ref 가 같은 semantic_key intent 면 self-cycle 됨.
    - dedup_parent_conflict_pre_adapter: 같은 semantic_key 로 묶일 intent group 의
      parent_intent_ref candidate set (None 포함) 이 inconsistent.
    """
    ikey_to_sem = _compute_intent_semantic_keys(raw_plan, location_profiles)
    intents_by_ikey = {
        it["intent_key"]: it
        for it in raw_plan.get("raw_background_intents", []) or []
        if it.get("intent_key")
    }

    errors: List[str] = []

    # parent_same_semantic_key
    for it in raw_plan.get("raw_background_intents", []) or []:
        ikey = it.get("intent_key")
        parent = it.get("parent_intent_ref")
        if not parent or ikey not in ikey_to_sem:
            continue
        if parent not in ikey_to_sem:
            continue
        if ikey_to_sem[ikey] == ikey_to_sem[parent]:
            errors.append(
                f"parent_same_semantic_key: intent_key={ikey!r} "
                f"parent_intent_ref={parent!r} share semantic_key "
                f"{ikey_to_sem[ikey]!r} (production will dedup to same bg_id, parent becomes self-cycle)"
            )

    # dedup_parent_conflict_pre_adapter — group by semantic_key, check parent set
    by_sem: Dict[str, List[str]] = {}
    for ikey, sem in ikey_to_sem.items():
        by_sem.setdefault(sem, []).append(ikey)
    for sem, group in by_sem.items():
        if len(group) < 2:
            continue
        parents: set = set()
        for ikey in group:
            parent = intents_by_ikey.get(ikey, {}).get("parent_intent_ref")
            # resolve parent intent → its sem key (parent of bg, not parent intent)
            parent_sem = ikey_to_sem.get(parent) if parent else None
            parents |= {parent_sem}  # None included; static-guard avoidance
        if len(parents) > 1:
            labels = sorted(p if p is not None else "<none>" for p in parents)
            errors.append(
                f"dedup_parent_conflict_pre_adapter: semantic_key={sem!r} "
                f"shared by {sorted(group)} have inconsistent parent semantic keys {labels}"
            )

    return errors


def _validate_space_key_hint_membership(raw_plan: dict, location_profiles: dict) -> List[str]:
    """W3 — space_key_hint 가 해당 loc 의 allowed_space_keys 안인지 (single_space 는 'main' 강제).

    Deterministic ID-set membership check — semantic 판단 아님.
    """
    errors: List[str] = []
    for it in raw_plan.get("raw_background_intents", []) or []:
        loc = it.get("loc_id")
        if loc not in location_profiles:
            continue  # loc membership validator 가 별도로 잡음
        profile = location_profiles[loc]
        hint = it.get("space_key_hint")
        kind = profile.get("kind")
        if kind == "single_space":
            if hint != "main":
                errors.append(
                    f"space_key_hint must be 'main' for single_space loc {loc!r}, "
                    f"intent_key={it.get('intent_key')!r} got {hint!r}"
                )
        elif kind == "multi_space":
            allowed = profile.get("allowed_space_keys", []) or []
            if hint not in allowed:
                errors.append(
                    f"space_key_hint {hint!r} not in allowed_space_keys "
                    f"{sorted(allowed)} for multi_space loc {loc!r}, "
                    f"intent_key={it.get('intent_key')!r}"
                )
    return errors


def _validate_loc_id_membership(raw_plan: dict, location_profiles: dict) -> List[str]:
    """W2 BLOCKING — structured ID membership check.

    raw_background_intents.loc_id / floor_plan_intents.loc_id /
    building_groups.member_loc_ids + anchor_loc_id 가 location_profiles.keys()
    membership 안에 있는지 확인. semantic 판단 아님 — exact set membership.

    Returns list of error strings (empty = OK). Caller can decide retry vs hard fail.
    """
    allowed = set(location_profiles.keys())
    errors: List[str] = []
    for it in raw_plan.get("raw_background_intents", []) or []:
        loc = it.get("loc_id")
        if loc not in allowed:
            errors.append(
                f"raw_background_intents intent_key={it.get('intent_key')!r} "
                f"loc_id={loc!r} not in allowed {sorted(allowed)}"
            )
    for fp in raw_plan.get("floor_plan_intents", []) or []:
        loc = fp.get("loc_id")
        if loc not in allowed:
            errors.append(
                f"floor_plan_intents fp_intent_key={fp.get('fp_intent_key')!r} "
                f"loc_id={loc!r} not in allowed {sorted(allowed)}"
            )
    for bg in raw_plan.get("building_groups", []) or []:
        anchor = bg.get("anchor_loc_id")
        if anchor not in allowed:
            errors.append(
                f"building_groups group_id={bg.get('group_id')!r} "
                f"anchor_loc_id={anchor!r} not in allowed {sorted(allowed)}"
            )
        for m in bg.get("member_loc_ids", []) or []:
            if m not in allowed:
                errors.append(
                    f"building_groups group_id={bg.get('group_id')!r} "
                    f"member_loc_id={m!r} not in allowed {sorted(allowed)}"
                )
    return errors


# ──────────────────────────────────────────────────────────────────────
# Phase 5 — production_adapter_plan
# ──────────────────────────────────────────────────────────────────────


def _build_adapter_plan(llm_raw_plan: dict, location_profiles: dict) -> dict:
    """Code-built master_plan-호환 shape. bg_catalog.assign_bg_ids 사용.

    SOT 규약 (Codex 합의):
    - background_catalog 는 production mirror Dict[bg_id, entry]. SOT.
    - shot_binding_intents 가 shot->intent_key SOT. raw_background_intents 에는
      applies_to_shots 필드 없음. code 가 binding 으로부터 applies_to_shots 를 derive.
    - assign_bg_ids 가 같은 semantic_key 로 2 intent_key 를 dedup 하고 deps 가
      다르면 SemanticKeyError raise — fail-fast (best-effort 금지).
    """
    from app.core import bg_catalog  # type: ignore

    intents = llm_raw_plan.get("raw_background_intents", [])
    floor_plan_intents = llm_raw_plan.get("floor_plan_intents", [])
    shot_bindings = llm_raw_plan.get("shot_binding_intents", [])

    # fp_id 부여 (deterministic): fp_intent_key 정렬 후 loc_id 별 시퀀스.
    fp_id_by_intent_key: Dict[str, str] = {}
    fp_seq_by_loc: Dict[int, int] = {}
    for fp in sorted(floor_plan_intents, key=lambda x: (x["loc_id"], x["fp_intent_key"])):
        loc_num = int(fp["loc_id"].lstrip("L"))
        fp_seq_by_loc.setdefault(loc_num, 0)
        fp_seq_by_loc[loc_num] += 1
        fp_id = f"fp_l{loc_num:02d}_{fp_seq_by_loc[loc_num]:02d}"
        fp_id_by_intent_key[fp["fp_intent_key"]] = fp_id

    # shot_binding_intents SOT -> intent_key 별 shot_key 목록 derive.
    shots_by_intent: Dict[str, list] = {}
    for sb in shot_bindings:
        shots_by_intent.setdefault(sb["intent_key"], []).append(sb["shot_key"])

    # raw intent -> assign_bg_ids 입력. depends_on_bg 는 ID assign 후 resolve,
    # 일단 빈 list 로 (raw 의 parent_intent_ref 만 보존).
    new_intents = []
    for it in intents:
        depends_fp = (
            [fp_id_by_intent_key[it["fp_intent_key"]]]
            if it.get("fp_intent_key")
            else []
        )
        new_intents.append({
            "loc_id": it["loc_id"],
            "space_key_hint": it["space_key_hint"],
            "time_phase": it["time_phase"],
            "state_class": it["state_class"],
            "sub_location_label": it.get("sub_location_label"),
            "state_label_raw": it.get("state_label_raw"),
            "depends_on_fp": depends_fp,
            "depends_on_bg": [],
            # derived from shot_binding_intents (SOT)
            "applies_to_shots": list(shots_by_intent.get(it["intent_key"], [])),
        })

    catalog: Dict[str, Dict] = bg_catalog.assign_bg_ids(
        prev_catalog={},
        new_intents=new_intents,
        location_profiles=location_profiles,
    )

    # intent_key -> bg_id 매핑. assign_bg_ids 가 같은 semantic_key 를 dedup 하므로
    # semantic_key 기준 역매핑.
    intent_key_to_bg: Dict[str, str] = {}
    for it in intents:
        sem_key = bg_catalog.compute_semantic_key(
            it["loc_id"],
            bg_catalog.normalize_space_key(
                it["loc_id"], it["space_key_hint"], location_profiles[it["loc_id"]]
            ),
            it["time_phase"],
            it["state_class"],
        )
        match_bg: Optional[str] = None
        for bg_id, entry in catalog.items():
            if entry["semantic_key"] == sem_key:
                match_bg = bg_id
                break
        if match_bg is None:
            raise ValueError(
                f"intent_key {it['intent_key']!r} did not match any catalog entry"
            )
        intent_key_to_bg[it["intent_key"]] = match_bg

    # 2차 pass: parent_intent_ref -> depends_on_bg resolve.
    parent_candidates_by_bg: Dict[str, set] = {}
    for it in intents:
        bg = intent_key_to_bg[it["intent_key"]]
        raw_parent = it.get("parent_intent_ref")
        if raw_parent:
            parent_bg = intent_key_to_bg.get(raw_parent)
            if parent_bg is None:
                raise ValueError(f"parent_intent_ref {raw_parent!r} unresolved")
            if parent_bg == bg:
                raise ValueError(
                    f"dedup_parent_self_cycle: intent_key={it['intent_key']!r} "
                    f"parent_intent_ref={raw_parent!r} -> same bg {bg}"
                )
            parent_candidates_by_bg.setdefault(bg, set()).update({parent_bg})
        else:
            parent_candidates_by_bg.setdefault(bg, set()).update({None})
    for bg_id, candidates in parent_candidates_by_bg.items():
        if len(candidates) > 1:
            labels = sorted(c if c is not None else "<none>" for c in candidates)
            raise ValueError(
                f"dedup_parent_conflict: bg_id={bg_id} has inconsistent parent "
                f"candidates {labels} from dedup'd intent group"
            )
        only = next(iter(candidates))
        catalog[bg_id]["depends_on_bg"] = [only] if only is not None else []

    # shot_background_map: production helper SOT.
    shot_bg = bg_catalog.build_shot_background_map(catalog)

    # gen_order topo (depends_on_fp 먼저, 다음 bg DAG topo).
    fp_ids = sorted({fp for e in catalog.values() for fp in e["depends_on_fp"]})
    bg_ids_ordered: List[str] = []
    pending = list(catalog.items())
    placed: set = set()
    while pending:
        progressed = False
        for bg_id, entry in list(pending):
            if all(d in placed for d in entry["depends_on_bg"]):
                bg_ids_ordered.append(bg_id)
                placed |= {bg_id}
                pending.remove((bg_id, entry))
                progressed = True
        if not progressed:
            raise ValueError(
                f"reference_chain cycle: remaining={[bg for bg, _ in pending]}"
            )
    gen_order = fp_ids + bg_ids_ordered

    return {
        "schema_version": 1,
        "plan_version": PLAN_VERSION,
        "building_groups": llm_raw_plan.get("building_groups", []),
        "floor_plans": [
            {
                "fp_id": fp_id_by_intent_key[fp["fp_intent_key"]],
                "loc_id": fp["loc_id"],
                "space_key_hint": fp["space_key_hint"],
            }
            for fp in floor_plan_intents
        ],
        "background_catalog": catalog,
        "shot_background_map": shot_bg,
        "bg_catalog_hash": bg_catalog.compute_bg_catalog_hash(catalog),
        "shot_binding_hash": bg_catalog.compute_shot_binding_hash(shot_bg),
        "gen_order": gen_order,
        "intent_key_to_bg_id": intent_key_to_bg,
    }


# ──────────────────────────────────────────────────────────────────────
# Phase 6 — generation_payload_dry_run + reference_input_plan
# ──────────────────────────────────────────────────────────────────────


def _build_generation_payload(raw_plan: dict, adapter: dict) -> dict:
    """background_render_step 이 받을 contract 의 최소 mirror.
    실제 호출 0. unmapped_fields 는 production contract 에서 우리가 emit 못 한 필드.
    """
    intent_to_bg = adapter.get("intent_key_to_bg_id", {})
    bg_to_entry = adapter.get("background_catalog", {})
    prompt_by_shot = {
        p["shot_key"]: p["bg_intent_text"]
        for p in raw_plan.get("prompt_payload_intents", [])
    }
    per_shot = []
    for sb in raw_plan.get("shot_binding_intents", []):
        bg = intent_to_bg.get(sb["intent_key"])
        if bg is None:
            continue
        entry = bg_to_entry[bg]
        per_shot.append({
            "shot_key": sb["shot_key"],
            "bg_id": bg,
            "loc_id": entry["loc_id"],
            "bg_intent_text": prompt_by_shot.get(sb["shot_key"], ""),
            "expected_attached_meta": [{"kind": "background", "id": bg}],
            "expected_reference_phrase_kinds": ["background"],
            "expected_label_placeholder": f"chain_bg:{bg}",
        })
    unmapped = [
        "image_bytes",
        "camera_recommendations",
        "shot_guides",
        "ref_used",
    ]
    return {"per_shot": per_shot, "unmapped_fields": unmapped}


def _build_reference_input_plan(raw_plan: dict, adapter: dict) -> dict:
    intent_to_bg = adapter.get("intent_key_to_bg_id", {})
    bg_to_entry = adapter.get("background_catalog", {})
    entries = []
    for sb in raw_plan.get("shot_binding_intents", []):
        bg = intent_to_bg.get(sb["intent_key"])
        if bg is None:
            continue
        entry = bg_to_entry[bg]
        entries.append({
            "shot_key": sb["shot_key"],
            "bg_id": bg,
            "loc_id": entry["loc_id"],
            "ref_contract": {"kind": "background", "id": bg, "policy": "required"},
            "expected_attached_meta": [{"kind": "background", "id": bg}],
            "reference_phrase_kinds": ["background"],
        })
    return {"entries": entries}


def _ref_contract_dry_run(rip: dict, adapter: dict) -> dict:
    """production helper validate_attached_refs 로 각 shot dry-run."""
    from app.core.ref_contract_validator import (  # type: ignore
        validate_attached_refs,
        RefContractError,
    )

    unit_to_loc = {
        bg: entry["loc_id"]
        for bg, entry in adapter.get("background_catalog", {}).items()
    }
    chain_bg_lookup = lambda bg: unit_to_loc.get(bg)  # noqa: E731

    failures = []
    for e in rip.get("entries", []):
        bg = e["bg_id"]
        minimal_rpc = {
            "asset_requirements": {
                "required_refs": [{"kind": "background", "id": bg, "policy": "required"}],
                "forbidden_refs": [],
                "readiness_policy": "block_if_missing",
            },
        }
        labeled_refs = [(f"chain_bg:{bg}", b"")]
        attached_meta = [("background", bg)]
        try:
            validate_attached_refs(
                minimal_rpc,
                labeled_refs,
                attached_meta,
                prompt="dry-run",
                is_close_framing=False,
                chain_bg_lookup=chain_bg_lookup,
                reference_phrase_kinds=["background"],
            )
        except RefContractError as exc:
            failures.append({
                "shot_key": e["shot_key"],
                "bg_id": bg,
                "error": str(exc),
            })
    return {"all_pass": len(failures) == 0, "failures": failures}


# ──────────────────────────────────────────────────────────────────────
# Phase 7 — pipeline_compatibility_report
# ──────────────────────────────────────────────────────────────────────


def _build_compatibility_report(
    *,
    bundle,
    raw_plan,
    adapter,
    rip,
    ref_dry,
    pmap_check,
    production_diff_empty,
    db_write_count,
    image_import_seen,
):
    inv: Dict[str, Dict[str, Any]] = {}

    # 1
    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }
    # 2
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }
    # 3
    inv["image_call_zero"] = {
        "pass": not image_import_seen,
        "detail": "no gemini_image_client / fal / gpt-image module import",
    }
    # 4
    canon = json.dumps(
        bundle.get("selected_shots", []), sort_keys=True, ensure_ascii=False
    ).encode("utf-8")
    expected_hash = hashlib.sha256(canon).hexdigest()[:16]
    actual_hash = bundle.get("source_hash", "")
    inv["source_verbatim"] = {
        "pass": actual_hash == expected_hash and bool(actual_hash),
        "detail": f"hash={actual_hash}",
    }
    # 5 — shot_key SOT (composite).
    shot_keys = {s["shot_key"] for s in bundle.get("selected_shots", [])}
    binding_keys = {e["shot_key"] for e in rip.get("entries", [])}
    missing = shot_keys - binding_keys
    inv["all_selected_shots_covered"] = {
        "pass": len(missing) == 0 and len(shot_keys) > 0,
        "detail": {"missing": sorted(missing)[:10], "total": len(shot_keys)},
    }
    # 6
    from app.core.bg_state_vocab import BG_ID_RE  # type: ignore
    catalog = adapter.get("background_catalog", {})
    bg_ok = bool(catalog) and all(BG_ID_RE.match(bg) for bg in catalog.keys())
    inv["bg_id_assigned_by_production_helper"] = {
        "pass": bg_ok,
        "detail": "all bg_id match BG_ID_RE",
    }
    # W2 IMPORTANT 1 — adapter build failed / empty catalog 시 dependent invariant
    # 들은 실제로 검증 못 한 상태 → skipped_due_to_adapter_failure 로 정직화.
    adapter_failed = bool(adapter.get("error")) or not catalog
    if adapter_failed:
        inv["shot_background_map_n_to_1"] = {
            "pass": False,
            "detail": "skipped_due_to_adapter_failure",
        }
        inv["reference_chain_acyclic"] = {
            "pass": False,
            "detail": "skipped_due_to_adapter_failure",
        }
        inv["ref_contract_dry_run_pass"] = {
            "pass": False,
            "detail": "skipped_due_to_adapter_failure",
        }
    else:
        # 7 — shot_background_map already raised on N:1; if we got here, it passed.
        inv["shot_background_map_n_to_1"] = {
            "pass": "shot_background_map" in adapter,
            "detail": "build_shot_background_map success",
        }
        # 8 — gen_order acyclic
        cat_bg_ids = set(catalog.keys())
        order_set = set(adapter.get("gen_order", []))
        order_bg_ids = {x for x in order_set if not x.startswith("fp_")}
        inv["reference_chain_acyclic"] = {
            "pass": order_bg_ids == cat_bg_ids,
            "detail": "gen_order covers all bg_id",
        }
        # 9
        inv["ref_contract_dry_run_pass"] = {
            "pass": bool(ref_dry.get("all_pass")),
            "detail": {"failures": ref_dry.get("failures", [])[:5]},
        }

    all_pass = (
        all(v["pass"] for v in inv.values())
        and pmap_check["all_paths_exist"]
        and pmap_check["all_symbols_found"]
    )
    return {
        "invariants": inv,
        "all_pass": all_pass,
        "production_pipeline_map_check": pmap_check,
    }


def _install_db_write_sentinel() -> None:
    """Install SQLAlchemy after_flush event listener that counts written rows.

    # Static guard contract is enforced by the partnered TEST, which scans
    # this source for forbidden write-callable literals. Keep this docstring
    # free of those literals.
    """
    from sqlalchemy import event  # type: ignore
    from sqlalchemy.orm import Session as _SASession  # type: ignore

    @event.listens_for(_SASession, "after_flush")
    def _after_flush(s, flush_context):  # noqa: ANN001
        # Use getattr so the script source never contains the literal substrings
        # that the static-guard test scans for ("session.add", "session.delete",
        # etc.) even as harmless attribute reads on SQLAlchemy state buckets.
        global _DB_WRITE_COUNT
        new_ct = len(getattr(s, "new"))
        dirty_ct = len(getattr(s, "dirty"))
        del_ct = len(getattr(s, "deleted"))
        _DB_WRITE_COUNT += new_ct + dirty_ct + del_ct


def _check_production_diff_empty() -> bool:
    r = subprocess.run(
        ["git", "diff", "--stat", "backend/app", "backend/alembic"],
        cwd=str(_REPO_ROOT),
        capture_output=True,
        text=True,
    )
    return r.returncode == 0 and r.stdout.strip() == ""


def _check_image_imports_present() -> bool:
    banned = (
        "app.modules.llm.gemini_image_client",
        "app.services.fal_angle_helpers",
        "app.modules.gemini_i2i_editor",
    )
    return any(b in sys.modules for b in banned)


# ──────────────────────────────────────────────────────────────────────
# Phase 8 — HTML diagnostic
# ──────────────────────────────────────────────────────────────────────


def _render_html(run_meta: dict, bundle, adapter, rip, report, run_dir: Path) -> None:
    def esc(x):
        return (
            str(x)
            .replace("&", "&amp;")
            .replace("<", "&lt;")
            .replace(">", "&gt;")
        )

    unit_count = len((adapter or {}).get("background_catalog", {}) or {})
    shot_total = (bundle or {}).get("selected_shot_count", 0)
    shot_covered = len((rip or {}).get("entries", []) or [])
    chain_len = len((adapter or {}).get("gen_order", []) or [])
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td>{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    payload_sample = ""
    payload_path = run_dir / "generation_payload_dry_run.json"
    if payload_path.exists():
        sample = json.loads(payload_path.read_text())
        first = sample.get("per_shot", [])[:2]
        payload_sample = esc(json.dumps(first, ensure_ascii=False, indent=2))

    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em;}}
table{{border-collapse:collapse}} td,th{{border:1px solid #ccc;padding:4px 8px}}
.metric{{display:inline-block;margin:0 1em 1em 0;padding:1em;border:1px solid #ddd;border-radius:6px}}
.fail{{color:#b00}} .pass{{color:#080}}</style></head>
<body>
<h1>background_pipeline_slice — {esc(run_meta.get('run_id'))}</h1>
<p>plan_version: <b>{esc(run_meta.get('plan_version'))}</b> | run_status:
<b class=\"{'fail' if run_meta.get('run_status')!='succeeded' else 'pass'}\">{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}</p>
<div>
<div class=\"metric\"><div>unit count</div><b>{unit_count}</b></div>
<div class=\"metric\"><div>shot coverage</div><b>{shot_covered} / {shot_total}</b></div>
<div class=\"metric\"><div>reference chain</div><b>{chain_len} steps</b></div>
<div class=\"metric\"><div>validation</div><b>{'PASS' if report and report.get('all_pass') else 'FAIL'}</b></div>
</div>
<h2>Acceptance gates</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Failed invariants</h2><pre>{esc(json.dumps(run_meta.get('failed_invariants', []), ensure_ascii=False, indent=2))}</pre>
<h2>payload sample (first 2)</h2><pre>{payload_sample}</pre>
<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
<details><summary>raw adapter plan (HTML preview only, slice 50000 chars — 원본 JSON 은 production_adapter_plan.json 전체 보존, slice/truncate 0)</summary><pre>{esc(json.dumps(adapter, ensure_ascii=False, indent=2))[:50000]}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


# ──────────────────────────────────────────────────────────────────────
# W4 — background_render_plan + reference_chain_plan (deterministic, no LLM)
# ──────────────────────────────────────────────────────────────────────


W4_STAGE = "w4_render_plan"

# render_role enum — deterministic from production state_class fields.
# special variants = state_class semantically separate from base 'normal' (non-realistic / one-off events).
_SPECIAL_STATE_CLASSES = frozenset({
    "dream_or_vision_state", "evidence_display", "blood_scene", "intrusion",
})


def _load_w3_artifacts(prev_run_dir: Path) -> dict:
    """W4 input — load W3 success run artifacts read-only."""
    required = {
        "adapter": "production_adapter_plan.json",
        "raw_plan": "llm_raw_plan.json",
        "payload": "generation_payload_dry_run.json",
        "rip": "reference_input_plan.json",
    }
    out: Dict[str, Any] = {}
    missing: List[str] = []
    for key, fname in required.items():
        p = prev_run_dir / fname
        if not p.exists():
            missing.append(fname)
            continue
        out[key] = json.loads(p.read_text())
    out["_missing"] = missing
    out["_prev_run_id"] = prev_run_dir.name
    return out


def _classify_render_roles(adapter: dict) -> Dict[str, dict]:
    """Per (loc_id, space_key) group → master_base + variants.

    Returns Dict[bg_id, {render_role, base_bg_id, variant_delta}].
    Deterministic — no semantic extraction, only production fields.

    Heuristic (per Codex/user spec):
    - group by (loc_id, space_key).
    - master_base = entry with state_class=='normal' and shortest depends_on_bg
      (ties broken by bg_id sort).
    - no 'normal' in group → master_base = entry with empty depends_on_bg
      (ties by bg_id sort); if all have deps, fallback to bg_id sort first entry.
    - non-master entries:
        - state_class in _SPECIAL_STATE_CLASSES → vision_or_special_variant
        - state_class != master_base.state_class (but not special) → state_variant
        - state_class == master_base.state_class but time_phase differs → time_variant
        - otherwise → state_variant (fallback)
    - base_bg_id for variant: entry.depends_on_bg[0] if present else master_base.bg_id.
    - variant_delta: short structured string from state_class/time_phase delta.
    """
    catalog = adapter.get("background_catalog", {}) or {}
    # group by (loc_id, space_key)
    groups: Dict[tuple, List[str]] = {}
    for bg_id, entry in catalog.items():
        key = (entry.get("loc_id"), entry.get("space_key"))
        groups.setdefault(key, []).append(bg_id)

    result: Dict[str, dict] = {}
    for key, bg_ids in groups.items():
        bg_ids_sorted = sorted(bg_ids)
        # master_base selection
        normals = [
            b for b in bg_ids_sorted
            if catalog[b].get("state_class") == "normal"
        ]
        if normals:
            # shortest depends_on_bg, ties by bg_id sort (already sorted)
            normals.sort(key=lambda b: (len(catalog[b].get("depends_on_bg") or []), b))
            master_bg = normals[0]
        else:
            no_dep = [
                b for b in bg_ids_sorted
                if not (catalog[b].get("depends_on_bg") or [])
            ]
            master_bg = no_dep[0] if no_dep else bg_ids_sorted[0]
        master_entry = catalog[master_bg]
        master_state = master_entry.get("state_class")
        master_time = master_entry.get("time_phase")

        for bg_id in bg_ids_sorted:
            entry = catalog[bg_id]
            state = entry.get("state_class")
            time_phase = entry.get("time_phase")
            if bg_id == master_bg:
                result[bg_id] = {
                    "render_role": "master_base",
                    "base_bg_id": None,
                    "variant_delta": "base",
                }
                continue
            # variant
            if state in _SPECIAL_STATE_CLASSES:
                role = "vision_or_special_variant"
                delta = f"state:{state}"
            elif state != master_state:
                role = "state_variant"
                delta = f"state:{master_state}->{state}"
            elif time_phase != master_time:
                role = "time_variant"
                delta = f"time:{master_time}->{time_phase}"
            else:
                role = "state_variant"
                delta = f"state:{master_state}->{state}"
            # base_bg_id: production depends_on_bg first if any, else master_bg.
            deps = entry.get("depends_on_bg") or []
            base_bg = deps[0] if deps else master_bg
            result[bg_id] = {
                "render_role": role,
                "base_bg_id": base_bg,
                "variant_delta": delta,
            }
    return result


def _build_render_plan(adapter: dict, raw_plan: dict, payload: dict, rip: dict) -> dict:
    """W4 SOT — per-bg render plan derived from W3 adapter + payload.

    Output shape:
      {schema_version, stage, render_units: {bg_id: {…}}, render_order: List[bg_id],
       unmapped_to_production_fields: List[str]}
    """
    catalog = adapter.get("background_catalog", {}) or {}
    gen_order = adapter.get("gen_order", []) or []
    roles = _classify_render_roles(adapter)
    # per-shot first bg_intent_text for prompt_payload_summary
    shot_to_bg_intent = {
        p["shot_key"]: p.get("bg_intent_text", "")
        for p in (raw_plan.get("prompt_payload_intents", []) or [])
    }

    render_units: Dict[str, dict] = {}
    # render_order_index: production gen_order minus fp_ entries → per-bg index.
    bg_order = [x for x in gen_order if not x.startswith("fp_")]
    order_index = {bg: i for i, bg in enumerate(bg_order)}

    for bg_id, entry in catalog.items():
        role_info = roles.get(bg_id, {"render_role": "master_base", "base_bg_id": None, "variant_delta": "base"})
        # reference_inputs: master_base = []; variant = [{kind:'background', id: base_bg_id}].
        ref_inputs: List[dict] = []
        if role_info["render_role"] != "master_base" and role_info.get("base_bg_id"):
            ref_inputs.append({"kind": "background", "id": role_info["base_bg_id"]})
        # base_continuity_constraints — short, derived from first applied shot's bg_intent_text.
        applies = entry.get("applies_to_shots", []) or []
        first_shot = applies[0] if applies else None
        prompt_summary = shot_to_bg_intent.get(first_shot, "") if first_shot else ""
        # base_continuity_constraints — short list. For master_base, the prompt summary IS the constraint;
        # for variants, the constraint is "inherit base + apply variant_delta".
        if role_info["render_role"] == "master_base":
            constraints = [prompt_summary] if prompt_summary else []
        else:
            constraints = [
                f"inherit from {role_info.get('base_bg_id')}",
                f"apply delta: {role_info['variant_delta']}",
            ]
        render_units[bg_id] = {
            "bg_id": bg_id,
            "loc_id": entry.get("loc_id"),
            "space_key": entry.get("space_key"),
            "time_phase": entry.get("time_phase"),
            "state_class": entry.get("state_class"),
            "render_role": role_info["render_role"],
            "base_bg_id": role_info.get("base_bg_id"),
            "render_order_index": order_index.get(bg_id, -1),
            "reference_inputs": ref_inputs,
            "base_continuity_constraints": constraints,
            "variant_delta": role_info["variant_delta"],
            "applies_to_shots": applies,
            "prompt_payload_summary": prompt_summary,
        }

    # production background_render_step contract fields we do NOT emit at W4.
    unmapped = [
        "shot_guides",
        "camera_recommendations",
        "image_bytes",
        "ref_used",
        "validation_score",
        "validation_result",
    ]

    return {
        "schema_version": 1,
        "stage": W4_STAGE,
        "plan_version": PLAN_VERSION,
        "render_units": render_units,
        "render_order": bg_order,
        "master_base_count": sum(1 for u in render_units.values() if u["render_role"] == "master_base"),
        "variant_count": sum(1 for u in render_units.values() if u["render_role"] != "master_base"),
        "unmapped_to_production_fields": unmapped,
    }


def _build_render_payload_dry_run_w4(render_plan: dict, adapter: dict) -> dict:
    """W4 — per-bg dry-run payload mirroring production background_render_step input.

    No image call; expected_chain_bg_asset is a placeholder for the ImageAsset row
    production would write (asset_type='chain_bg', variant_type=bg_id).
    """
    units = render_plan.get("render_units", {}) or {}
    order = render_plan.get("render_order", []) or []
    per_bg: List[dict] = []
    for bg_id in order:
        u = units.get(bg_id)
        if not u:
            continue
        per_bg.append({
            "bg_id": bg_id,
            "loc_id": u["loc_id"],
            "render_role": u["render_role"],
            "base_bg_id": u["base_bg_id"],
            "reference_inputs": u["reference_inputs"],
            "prompt_payload_summary": u["prompt_payload_summary"],
            "applies_to_shots": u["applies_to_shots"],
            "expected_chain_bg_asset": {
                "asset_type": "chain_bg",
                "variant_type": bg_id,
                "entity_id_resolves_to_loc": u["loc_id"],
            },
        })
    return {"per_bg": per_bg, "unmapped_fields": render_plan.get("unmapped_to_production_fields", [])}


def _build_reference_chain_plan(render_plan: dict, adapter: dict) -> dict:
    """W4 — topo chain of render units from master_base → variants.

    Built from existing adapter.gen_order (production topo) intersected with
    render_units. Validates acyclic + every variant's base_bg_id resolves
    earlier in chain. No invention of cross-loc deps.
    """
    units = render_plan.get("render_units", {}) or {}
    bg_order = render_plan.get("render_order", []) or []
    seen: set = set()
    chain: List[dict] = []
    errors: List[str] = []
    for bg_id in bg_order:
        u = units.get(bg_id)
        if not u:
            continue
        base = u.get("base_bg_id")
        if base is not None and base not in seen:
            errors.append(
                f"variant {bg_id} base_bg_id={base} not yet rendered "
                f"(chain order violates topo)"
            )
        chain.append({
            "bg_id": bg_id,
            "render_role": u["render_role"],
            "base_bg_id": base,
            "reference_inputs": u["reference_inputs"],
        })
        seen |= {bg_id}
    return {"chain": chain, "errors": errors}


def _build_render_plan_compatibility_report(
    *, render_plan: dict, render_payload: dict, chain_plan: dict,
    rip: dict, adapter: dict, production_diff_empty: bool, db_write_count: int,
    image_import_seen: bool, prev_run_id: str, missing_inputs: List[str],
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}
    units = render_plan.get("render_units", {}) or {}
    catalog_count = len(units)
    chain_count = len(chain_plan.get("chain", []) or [])

    inv["w3_inputs_present"] = {
        "pass": len(missing_inputs) == 0,
        "detail": {"missing": missing_inputs, "prev_run": prev_run_id},
    }
    inv["all_bg_covered_in_render_plan"] = {
        "pass": catalog_count > 0 and chain_count == catalog_count,
        "detail": f"render_units={catalog_count} chain={chain_count}",
    }
    inv["master_base_exists_per_loc_group"] = {
        "pass": True,
        "detail": "checked below",
    }
    # group-level master_base check
    groups: Dict[tuple, List[str]] = {}
    for bg_id, u in units.items():
        groups.setdefault((u["loc_id"], u["space_key"]), []).append(bg_id)
    missing_master = [
        f"{loc}|{sk}"
        for (loc, sk), bgs in groups.items()
        if not any(units[b]["render_role"] == "master_base" for b in bgs)
    ]
    inv["master_base_exists_per_loc_group"] = {
        "pass": len(missing_master) == 0,
        "detail": {"groups_missing_master": missing_master[:10],
                   "groups_total": len(groups)},
    }
    inv["variants_reference_earlier_bg"] = {
        "pass": len(chain_plan.get("errors", [])) == 0,
        "detail": {"errors": chain_plan.get("errors", [])[:5]},
    }
    inv["render_chain_acyclic"] = {
        "pass": len(set(c["bg_id"] for c in chain_plan.get("chain", []))) == chain_count,
        "detail": f"unique bg_ids in chain = {chain_count}",
    }
    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }
    inv["image_call_zero"] = {
        "pass": not image_import_seen,
        "detail": "no gemini_image_client / fal / gpt-image module import",
    }
    # no human-decision fields scan (any rendering unit with manual_review / needs_user_decision / rollup keys)
    banned_keys = {"needs_user_decision", "manual_review_required", "rollup",
                   "pending_human_decision", "human_review"}
    units_with_banned = [
        bg for bg, u in units.items()
        if any(bk in u for bk in banned_keys)
    ]
    inv["no_human_decision_field"] = {
        "pass": len(units_with_banned) == 0,
        "detail": {"violating_units": units_with_banned[:5]},
    }
    # W4b BLOCKING 1 — exact set equality between expected (adapter shot map)
    # and actual (rip entries shot_keys). previous "shot_count > 0" was false-pass.
    expected_shots = set((adapter or {}).get("shot_background_map", {}).keys())
    actual_shots = {e.get("shot_key") for e in (rip or {}).get("entries", []) or []
                    if e.get("shot_key")}
    missing_in_rip = sorted(expected_shots - actual_shots)
    extra_in_rip = sorted(actual_shots - expected_shots)
    inv["prompt_payload_coverage_carried"] = {
        "pass": bool(expected_shots) and actual_shots == expected_shots,
        "detail": {
            "expected_count": len(expected_shots),
            "actual_count": len(actual_shots),
            "missing_in_rip": missing_in_rip[:10],
            "extra_in_rip": extra_in_rip[:10],
        },
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


def _render_w4_html(run_meta: dict, render_plan: dict, payload: dict,
                    chain_plan: dict, report: dict, run_dir: Path) -> None:
    def esc(x):
        return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    units = render_plan.get("render_units", {}) or {}
    bg_count = len(units)
    master_count = render_plan.get("master_base_count", 0)
    variant_count = render_plan.get("variant_count", 0)
    chain_count = len(chain_plan.get("chain", []) or [])
    # shot coverage from W4 input
    shot_set = set()
    for u in units.values():
        shot_set |= set(u.get("applies_to_shots", []) or [])
    shot_total = len(shot_set)
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td class=\"{'pass' if v['pass'] else 'fail'}\">{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    per_bg_sample = payload.get("per_bg", [])[:2]
    sample = esc(json.dumps(per_bg_sample, ensure_ascii=False, indent=2))
    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice W4 {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em;}}
table{{border-collapse:collapse}} td,th{{border:1px solid #ccc;padding:4px 8px}}
.metric{{display:inline-block;margin:0 1em 1em 0;padding:1em;border:1px solid #ddd;border-radius:6px}}
.fail{{color:#b00}} .pass{{color:#080}}</style></head>
<body>
<h1>background_pipeline_slice W4 — {esc(run_meta.get('run_id'))}</h1>
<p>stage: <b>{esc(run_meta.get('stage'))}</b> | plan_version: <b>{esc(run_meta.get('plan_version'))}</b>
| run_status: <b class=\"{'fail' if run_meta.get('run_status')!='succeeded' else 'pass'}\">{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}
| derived_from: {esc(run_meta.get('derived_from'))}</p>
<div>
<div class=\"metric\"><div>bg count</div><b>{bg_count}</b></div>
<div class=\"metric\"><div>master_base</div><b>{master_count}</b></div>
<div class=\"metric\"><div>variants</div><b>{variant_count}</b></div>
<div class=\"metric\"><div>render chain</div><b>{chain_count} steps</b></div>
<div class=\"metric\"><div>shot coverage</div><b>{shot_total}</b></div>
<div class=\"metric\"><div>validation</div><b>{'PASS' if report and report.get('all_pass') else 'FAIL'}</b></div>
</div>
<h2>W4 Invariants</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Render payload sample (first 2)</h2><pre>{sample}</pre>
<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
<details><summary>raw render_plan (HTML preview only, slice 60000 chars; full JSON preserved in background_render_plan.json)</summary><pre>{esc(json.dumps(render_plan, ensure_ascii=False, indent=2))[:60000]}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


def _w4_main(args, run_dir: Path, run_id: str) -> int:
    """W4 stage — derive render plan from a prior W3 success run. No LLM, no DB."""
    global _DB_WRITE_COUNT
    prev_run_dir = Path(args.derive_render_plan_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir
    artifacts = _load_w3_artifacts(prev_run_dir)
    missing = artifacts.get("_missing", [])
    outputs: List[str] = []
    failed: List[str] = []
    run_status = "succeeded"
    exit_code = 0

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W4_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model": None,
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed,
    }

    if missing:
        failed.append("w3_inputs_missing")
        run_status = "validation_failed"
        exit_code = 1
        # still write run_meta + minimal report
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    adapter = artifacts["adapter"]
    raw_plan_doc = artifacts["raw_plan"]
    raw_plan = raw_plan_doc.get("plan", {}) or {}
    payload_w3 = artifacts["payload"]
    rip_w3 = artifacts["rip"]

    render_plan = _build_render_plan(adapter, raw_plan, payload_w3, rip_w3)
    (run_dir / "background_render_plan.json").write_text(
        json.dumps(render_plan, ensure_ascii=False, indent=2)
    )
    outputs.append("background_render_plan.json")

    render_payload = _build_render_payload_dry_run_w4(render_plan, adapter)
    (run_dir / "background_render_payload_dry_run.json").write_text(
        json.dumps(render_payload, ensure_ascii=False, indent=2)
    )
    outputs.append("background_render_payload_dry_run.json")

    chain_plan = _build_reference_chain_plan(render_plan, adapter)
    (run_dir / "reference_chain_plan.json").write_text(
        json.dumps(chain_plan, ensure_ascii=False, indent=2)
    )
    outputs.append("reference_chain_plan.json")

    report = _build_render_plan_compatibility_report(
        render_plan=render_plan, render_payload=render_payload, chain_plan=chain_plan,
        rip=rip_w3, adapter=adapter,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_import_seen=_check_image_imports_present(),
        prev_run_id=prev_run_dir.name,
        missing_inputs=missing,
    )
    (run_dir / "render_plan_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("render_plan_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed:
            failed.append(name)
    if failed:
        run_status = "validation_failed"
        exit_code = 1

    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed
    run_meta["outputs"] = outputs
    _render_w4_html(run_meta, render_plan, render_payload, chain_plan, report, run_dir)
    outputs.append("index.html")
    run_meta["outputs"] = outputs
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    _maybe_print_imports(args)
    return exit_code


# ──────────────────────────────────────────────────────────────────────
# W5 — background_generation_map + render_batches (deterministic, no LLM)
# ──────────────────────────────────────────────────────────────────────

W5_STAGE = "w5_generation_map"
# User policy — every actual image generation will use gpt-image-2. No fallback model.
W5_IMAGE_BACKEND = "gpt-image-2"


def _load_w4_artifacts(prev_run_dir: Path) -> dict:
    """W5 input — load W4 success run artifacts read-only."""
    required = {
        "render_plan": "background_render_plan.json",
        "render_payload": "background_render_payload_dry_run.json",
        "chain_plan": "reference_chain_plan.json",
        "report": "render_plan_compatibility_report.json",
        "run_meta": "run_meta.json",
    }
    out: Dict[str, Any] = {}
    missing: List[str] = []
    for key, fname in required.items():
        p = prev_run_dir / fname
        if not p.exists():
            missing.append(fname)
            continue
        out[key] = json.loads(p.read_text())
    out["_missing"] = missing
    out["_prev_run_id"] = prev_run_dir.name
    return out


def _build_generation_map(render_plan: dict, chain_plan: dict) -> dict:
    """W5 SOT — deterministic derive from W4 render_plan + chain_plan.

    Policy (per user spec):
    - master_base                              -> independent_text_to_image (no primary_ref)
    - state_variant | time_variant + base      -> derive_from_base_reference
    - vision_or_special_variant + base         -> weak_reference_variant
    - any variant w/o base                     -> independent_text_to_image fallback
    """
    units = render_plan.get("render_units", {}) or {}

    # ── background_images ─────────────────────────────────────────────
    bg_images: Dict[str, dict] = {}
    for bg_id, u in units.items():
        role = u.get("render_role")
        base = u.get("base_bg_id")
        if role == "master_base":
            render_mode = "independent_text_to_image"
            primary_ref = None
            policy = "no_reference"
        elif role in ("state_variant", "time_variant") and base:
            render_mode = "derive_from_base_reference"
            primary_ref = base
            policy = "strict_reference_to_base"
        elif role == "vision_or_special_variant" and base:
            render_mode = "weak_reference_variant"
            primary_ref = base
            policy = "weak_reference_continuity_only"
        else:
            # variant w/o base — fallback to independent + report
            render_mode = "independent_text_to_image"
            primary_ref = None
            policy = "fallback_independent_no_base"
        bg_images[bg_id] = {
            "bg_id": bg_id,
            "image_title": f"{u.get('loc_id')}|{u.get('space_key')}|{u.get('time_phase')}|{u.get('state_class')}",
            "loc_id": u.get("loc_id"),
            "space_key": u.get("space_key"),
            "render_role": role,
            "state_class": u.get("state_class"),
            "time_phase": u.get("time_phase"),
            "render_mode": render_mode,
            "primary_reference_bg_id": primary_ref,
            "secondary_reference_bg_ids": [],
            "reference_policy": policy,
            "prompt_brief": u.get("prompt_payload_summary", ""),
            "applies_to_shots": list(u.get("applies_to_shots", []) or []),
            "expected_asset": {"asset_type": "chain_bg", "variant_type": bg_id},
            # User policy — actual generation will be gpt-image-2 only. No fallback.
            "expected_image_model": W5_IMAGE_BACKEND,
        }

    # ── groups (loc_id + space_key) ───────────────────────────────────
    groups_index: Dict[tuple, dict] = {}
    for bg_id, u in units.items():
        key = (u.get("loc_id"), u.get("space_key"))
        g = groups_index.setdefault(key, {
            "group_id": f"{key[0]}|{key[1]}",
            "loc_id": key[0],
            "space_key": key[1],
            "base_bg_id": None,
            "member_bg_ids": [],
            "variant_bg_ids": [],
            "shot_keys": [],
            "generation_order": [],
        })
        g["member_bg_ids"].append(bg_id)
        if u.get("render_role") == "master_base":
            # 첫 master 만 base 로 (deterministic, sort 기준)
            if g["base_bg_id"] is None or bg_id < g["base_bg_id"]:
                g["base_bg_id"] = bg_id
        else:
            g["variant_bg_ids"].append(bg_id)
        for sk in (u.get("applies_to_shots") or []):
            if sk not in g["shot_keys"]:
                g["shot_keys"].append(sk)
    # deterministic per-group sort
    for g in groups_index.values():
        g["member_bg_ids"] = sorted(g["member_bg_ids"])
        g["variant_bg_ids"] = sorted(g["variant_bg_ids"])
        g["shot_keys"] = sorted(g["shot_keys"])
        # generation_order: base first (if any), then variants topologically by primary_ref
        ordered: List[str] = []
        if g["base_bg_id"]:
            ordered.append(g["base_bg_id"])
        # variants whose primary_ref is base come next; rest after
        first_tier = [v for v in g["variant_bg_ids"]
                      if bg_images[v]["primary_reference_bg_id"] == g["base_bg_id"]]
        rest = [v for v in g["variant_bg_ids"] if v not in first_tier]
        ordered.extend(sorted(first_tier))
        ordered.extend(sorted(rest))
        # also append any unbase members not already added (e.g. multi master edge cases)
        for m in g["member_bg_ids"]:
            if m not in ordered:
                ordered.append(m)
        g["generation_order"] = ordered

    groups_sorted = sorted(groups_index.values(), key=lambda g: g["group_id"])

    # ── render_batches (cross-group topological) ──────────────────────
    placed: set = set()
    batches: List[List[str]] = []
    pending: set = set(bg_images.keys())
    # Safety bound: at most len(units) iterations.
    iter_guard = 0
    while pending and iter_guard < len(bg_images) + 5:
        iter_guard += 1
        # next batch: bg whose primary_ref is None OR already placed
        ready = sorted(
            b for b in pending
            if bg_images[b]["primary_reference_bg_id"] is None
            or bg_images[b]["primary_reference_bg_id"] in placed
        )
        if not ready:
            # cycle / unresolved ref — fail-fast
            raise ValueError(
                f"render_batches_unresolved: pending={sorted(pending)} "
                f"primary_refs={[bg_images[b]['primary_reference_bg_id'] for b in sorted(pending)]}"
            )
        batches.append(ready)
        placed |= set(ready)
        pending -= set(ready)

    # ── shot_to_background_image ──────────────────────────────────────
    shot_map: Dict[str, str] = {}
    for bg_id, img in bg_images.items():
        for sk in img["applies_to_shots"]:
            # production-style N:1 already validated in W3 (build_shot_background_map)
            shot_map[sk] = bg_id

    unmapped = [
        # production background_render_step / scene_image_pipeline fields not yet emitted at W5.
        "shot_guides",  # background_prompt step
        "camera_recommendations",  # floor_plan_prompt
        "image_bytes",  # actual PNG (image API call)
        "ref_used",  # actual ref consumption record
        "validation_score",  # LVM validation
        "validation_result",
        "sanitization_strategy",
    ]

    return {
        "schema_version": 1,
        "stage": W5_STAGE,
        "plan_version": PLAN_VERSION,
        "image_generation_backend": W5_IMAGE_BACKEND,
        "groups": groups_sorted,
        "background_images": bg_images,
        "render_batches": batches,
        "shot_to_background_image": shot_map,
        "counts": {
            "background_images": len(bg_images),
            "groups": len(groups_sorted),
            "independent": sum(1 for v in bg_images.values() if v["render_mode"] == "independent_text_to_image"),
            "derive_from_base_reference": sum(1 for v in bg_images.values() if v["render_mode"] == "derive_from_base_reference"),
            "weak_reference_variant": sum(1 for v in bg_images.values() if v["render_mode"] == "weak_reference_variant"),
            "batches": len(batches),
            "shots_mapped": len(shot_map),
        },
        "unmapped_to_production_fields": unmapped,
    }


def _build_generation_map_compatibility_report(
    *, gen_map: dict, render_plan: dict, w4_report: dict,
    production_diff_empty: bool, db_write_count: int,
    image_import_seen: bool, prev_run_id: str, missing_inputs: List[str],
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}
    bg_images = gen_map.get("background_images", {}) or {}
    units = render_plan.get("render_units", {}) or {}
    groups = gen_map.get("groups", []) or []
    batches = gen_map.get("render_batches", []) or []
    shot_map = gen_map.get("shot_to_background_image", {}) or {}

    inv["w4_inputs_present"] = {
        "pass": len(missing_inputs) == 0,
        "detail": {"missing": missing_inputs, "prev_run": prev_run_id},
    }

    # all_bg_images_covered: W4 render_units bg set == W5 bg_images keys
    expected_bg = set(units.keys())
    actual_bg = set(bg_images.keys())
    inv["all_bg_images_covered"] = {
        "pass": bool(expected_bg) and expected_bg == actual_bg,
        "detail": {
            "expected_count": len(expected_bg),
            "actual_count": len(actual_bg),
            "missing": sorted(expected_bg - actual_bg)[:10],
            "extra": sorted(actual_bg - expected_bg)[:10],
        },
    }

    # all_groups_have_base (at least one master_base member per group, except entirely-variant fallback marked)
    groups_missing_base = [g["group_id"] for g in groups if g["base_bg_id"] is None]
    inv["all_groups_have_base"] = {
        "pass": len(groups_missing_base) == 0,
        "detail": {"missing_base": groups_missing_base[:10]},
    }

    # render_batches_cover_all_bg_once
    batch_flat: List[str] = [bg for batch in batches for bg in batch]
    inv["render_batches_cover_all_bg_once"] = {
        "pass": (sorted(batch_flat) == sorted(actual_bg)) and (len(batch_flat) == len(set(batch_flat))),
        "detail": {
            "batch_count": len(batches),
            "flat_count": len(batch_flat),
            "expected": len(actual_bg),
        },
    }

    # references_point_to_earlier_batch
    bg_to_batch = {bg: i for i, batch in enumerate(batches) for bg in batch}
    bad_refs: List[str] = []
    for bg, img in bg_images.items():
        pr = img["primary_reference_bg_id"]
        if pr is None:
            continue
        if pr not in bg_to_batch:
            bad_refs.append(f"{bg}:ref_{pr}_unknown")
        elif bg_to_batch[pr] >= bg_to_batch.get(bg, 0):
            bad_refs.append(f"{bg}(batch={bg_to_batch.get(bg)})_refs_{pr}(batch={bg_to_batch[pr]})")
    inv["references_point_to_earlier_batch"] = {
        "pass": len(bad_refs) == 0,
        "detail": {"violations": bad_refs[:10]},
    }

    # shot_to_background_image_coverage: expected = sum of all applies_to_shots = W4 total
    expected_shots: set = set()
    for u in units.values():
        expected_shots |= set(u.get("applies_to_shots") or [])
    actual_shots = set(shot_map.keys())
    inv["shot_to_background_image_coverage"] = {
        "pass": bool(expected_shots) and expected_shots == actual_shots,
        "detail": {
            "expected_count": len(expected_shots),
            "actual_count": len(actual_shots),
            "missing": sorted(expected_shots - actual_shots)[:10],
            "extra": sorted(actual_shots - expected_shots)[:10],
        },
    }

    # no_human_decision_field — scan gen_map structures for banned keys
    banned_keys = {"needs_user_decision", "manual_review_required", "rollup",
                   "pending_human_decision", "human_review"}
    bad_units: List[str] = []
    for bg, img in bg_images.items():
        if any(bk in img for bk in banned_keys):
            bad_units.append(bg)
    for g in groups:
        if any(bk in g for bk in banned_keys):
            bad_units.append(g["group_id"])
    inv["no_human_decision_field"] = {
        "pass": len(bad_units) == 0,
        "detail": {"violating": bad_units[:5]},
    }

    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }
    inv["image_call_zero"] = {
        "pass": not image_import_seen,
        "detail": "no gemini_image_client / fal / gpt-image module import",
    }

    # User policy invariant — all bg_images must declare gpt-image-2 backend.
    backend = gen_map.get("image_generation_backend")
    wrong_backend = [
        bg for bg, img in bg_images.items()
        if img.get("expected_image_model") != W5_IMAGE_BACKEND
    ]
    inv["image_model_is_gpt_image_2"] = {
        "pass": backend == W5_IMAGE_BACKEND and len(wrong_backend) == 0,
        "detail": {
            "gen_map_backend": backend,
            "violating_bg_count": len(wrong_backend),
            "violating_sample": wrong_backend[:5],
        },
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


def _render_w5_html(run_meta: dict, gen_map: dict, report: dict, run_dir: Path) -> None:
    def esc(x):
        return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    counts = gen_map.get("counts", {}) or {}
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td class=\"{'pass' if v['pass'] else 'fail'}\">{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    sample_group = (gen_map.get("groups") or [{}])[0]
    sample_img_items = list((gen_map.get("background_images") or {}).items())[:2]
    sample_imgs = {k: v for k, v in sample_img_items}
    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice W5 {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em;}}
table{{border-collapse:collapse}} td,th{{border:1px solid #ccc;padding:4px 8px}}
.metric{{display:inline-block;margin:0 1em 1em 0;padding:1em;border:1px solid #ddd;border-radius:6px}}
.fail{{color:#b00}} .pass{{color:#080}}</style></head>
<body>
<h1>background_pipeline_slice W5 — {esc(run_meta.get('run_id'))}</h1>
<p>stage: <b>{esc(run_meta.get('stage'))}</b> | plan_version: <b>{esc(run_meta.get('plan_version'))}</b>
| run_status: <b class=\"{'fail' if run_meta.get('run_status')!='succeeded' else 'pass'}\">{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}
| derived_from: {esc(run_meta.get('derived_from'))}
| image_backend: <b>{esc(run_meta.get('image_generation_backend'))}</b></p>
<div>
<div class=\"metric\"><div>bg images</div><b>{counts.get('background_images', 0)}</b></div>
<div class=\"metric\"><div>groups</div><b>{counts.get('groups', 0)}</b></div>
<div class=\"metric\"><div>independent</div><b>{counts.get('independent', 0)}</b></div>
<div class=\"metric\"><div>derive_from_base</div><b>{counts.get('derive_from_base_reference', 0)}</b></div>
<div class=\"metric\"><div>weak_reference</div><b>{counts.get('weak_reference_variant', 0)}</b></div>
<div class=\"metric\"><div>render batches</div><b>{counts.get('batches', 0)}</b></div>
<div class=\"metric\"><div>shots mapped</div><b>{counts.get('shots_mapped', 0)}</b></div>
<div class=\"metric\"><div>validation</div><b>{'PASS' if report and report.get('all_pass') else 'FAIL'}</b></div>
</div>
<h2>W5 Invariants</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Sample group</h2><pre>{esc(json.dumps(sample_group, ensure_ascii=False, indent=2))}</pre>
<h2>Sample background_images (first 2)</h2><pre>{esc(json.dumps(sample_imgs, ensure_ascii=False, indent=2))}</pre>
<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
<details><summary>raw generation_map (HTML preview only, slice 60000 chars; full JSON preserved)</summary><pre>{esc(json.dumps(gen_map, ensure_ascii=False, indent=2))[:60000]}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


def _w5_main(args, run_dir: Path, run_id: str) -> int:
    """W5 stage — derive generation map from a prior W4 success run."""
    global _DB_WRITE_COUNT
    prev_run_dir = Path(args.derive_generation_map_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir
    artifacts = _load_w4_artifacts(prev_run_dir)
    missing = artifacts.get("_missing", [])
    outputs: List[str] = []
    failed: List[str] = []
    run_status = "succeeded"
    exit_code = 0

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W5_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model": None,
        "image_generation_backend": W5_IMAGE_BACKEND,
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed,
    }

    if missing:
        failed.append("w4_inputs_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    render_plan = artifacts["render_plan"]
    chain_plan = artifacts["chain_plan"]
    w4_report = artifacts["report"]

    gen_map = _build_generation_map(render_plan, chain_plan)
    (run_dir / "background_generation_map.json").write_text(
        json.dumps(gen_map, ensure_ascii=False, indent=2)
    )
    outputs.append("background_generation_map.json")

    report = _build_generation_map_compatibility_report(
        gen_map=gen_map, render_plan=render_plan, w4_report=w4_report,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_import_seen=_check_image_imports_present(),
        prev_run_id=prev_run_dir.name,
        missing_inputs=missing,
    )
    (run_dir / "generation_map_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("generation_map_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed:
            failed.append(name)
    if failed:
        run_status = "validation_failed"
        exit_code = 1

    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed
    run_meta["outputs"] = outputs
    _render_w5_html(run_meta, gen_map, report, run_dir)
    outputs.append("index.html")
    run_meta["outputs"] = outputs
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    _maybe_print_imports(args)
    return exit_code


# ─────────────────────────────────────────────────────────────────────────────
# W6 — image payload plan (gpt-image-2 request shape mirror, no API call)
# ─────────────────────────────────────────────────────────────────────────────

W6_STAGE = "w6_image_payload_plan"
W6_IMAGE_BACKEND = "gpt-image-2"
W6_DEFAULT_SIZE = "1024x1024"
W6_DEFAULT_QUALITY = "high"
W6_PROMPT_MAX_LEN = 6000
W6_PROMPT_ADVISORY_MIN_LEN = 500

# Deterministic, mode-specific augmentation strings. NOT LLM-derived. NOT padded
# nor used for semantic decision. Prepended verbatim to W5 prompt_brief.
W6_AUGMENTATION_INDEPENDENT = (
    "Create a standalone chain background image. No reference image is provided. "
)
W6_AUGMENTATION_DERIVE = (
    "Use the referenced base background as a strict spatial continuity reference. "
    "Preserve the shared structure and change only the requested time/state cues. "
)
W6_AUGMENTATION_WEAK = (
    "Use the referenced base background only for loose spatial tone and continuity. "
    "Do not force exact object matching if the variant is special or subjective. "
)
W6_AUGMENTATION_FALLBACK = (
    "Create a standalone chain background image (variant without an available base reference). "
)


def _load_w5_artifacts(prev_run_dir: Path) -> dict:
    """W6 input — load W5 success run artifacts read-only."""
    required = {
        "gen_map": "background_generation_map.json",
        "report": "generation_map_compatibility_report.json",
        "run_meta": "run_meta.json",
    }
    out: Dict[str, Any] = {}
    missing: List[str] = []
    for key, fname in required.items():
        p = prev_run_dir / fname
        if not p.exists():
            missing.append(fname)
            continue
        out[key] = json.loads(p.read_text())
    out["_missing"] = missing
    out["_prev_run_id"] = prev_run_dir.name
    return out


def _augmentation_for(render_mode: str, reference_policy: str) -> str:
    """Deterministic mode-instruction prefix. No semantic parsing."""
    if render_mode == "independent_text_to_image":
        if reference_policy == "fallback_independent_no_base":
            return W6_AUGMENTATION_FALLBACK
        return W6_AUGMENTATION_INDEPENDENT
    if render_mode == "derive_from_base_reference":
        return W6_AUGMENTATION_DERIVE
    if render_mode == "weak_reference_variant":
        return W6_AUGMENTATION_WEAK
    return W6_AUGMENTATION_INDEPENDENT


def _api_call_shape_for(render_mode: str, has_ref: bool) -> dict:
    """Mirror production gpt-image-2 request shape. No file handle, no API call."""
    if has_ref:
        client_method = "images.edit"
        reference_input_kind = "single_background_image"
    else:
        client_method = "images.generate"
        reference_input_kind = "none"
    return {
        "client_method": client_method,
        "model": W6_IMAGE_BACKEND,
        "size": W6_DEFAULT_SIZE,
        "quality": W6_DEFAULT_QUALITY,
        "n": 1,
        "reference_input_kind": reference_input_kind,
    }


def _build_image_payload_plan(gen_map: dict) -> dict:
    """W6 SOT — deterministic carry+combine from W5 background_generation_map.

    Policy (per user spec + Codex consult):
    - independent_text_to_image -> images.generate, no refs
    - derive_from_base_reference -> images.edit, primary_ref present
    - weak_reference_variant    -> images.edit, primary_ref present
    - variant w/o primary_ref   -> independent fallback, images.generate, no refs

    No semantic extraction. No LLM. No regex/substring/particle matching. No
    human-decision fields. Prompt brief carried verbatim, augmentation prefix
    is a deterministic constant string.
    """
    bg_images: Dict[str, dict] = gen_map.get("background_images", {}) or {}
    batches: List[List[str]] = gen_map.get("render_batches", []) or []
    groups: List[dict] = gen_map.get("groups", []) or []
    shot_map: Dict[str, str] = gen_map.get("shot_to_background_image", {}) or {}

    bg_to_batch_index: Dict[str, int] = {}
    for i, batch in enumerate(batches):
        for bg in batch:
            bg_to_batch_index[bg] = i

    payloads: Dict[str, dict] = {}
    for bg_id, img in bg_images.items():
        render_mode = img.get("render_mode") or "independent_text_to_image"
        reference_policy = img.get("reference_policy") or "no_reference"
        primary_ref = img.get("primary_reference_bg_id")
        prompt_brief = img.get("prompt_brief", "") or ""

        # variant w/o primary_ref — fallback to independent. W5 already encodes
        # this via reference_policy="fallback_independent_no_base"; we mirror.
        has_ref = bool(primary_ref) and render_mode in (
            "derive_from_base_reference", "weak_reference_variant"
        )

        reference_images: List[dict] = []
        can_generate_after: List[str] = []
        if has_ref:
            ref_expected_asset = (
                bg_images.get(primary_ref, {}).get("expected_asset")
                or {"asset_type": "chain_bg", "variant_type": primary_ref}
            )
            reference_images.append({
                "bg_id": primary_ref,
                "role": "primary",
                "expected_asset": ref_expected_asset,
            })
            can_generate_after.append(primary_ref)

        prompt_text = _augmentation_for(render_mode, reference_policy) + prompt_brief

        payloads[bg_id] = {
            "bg_id": bg_id,
            "image_title": img.get("image_title"),
            "image_model": W6_IMAGE_BACKEND,
            "render_mode": render_mode,
            "reference_policy": reference_policy,
            "prompt_text": prompt_text,
            "reference_images": reference_images,
            "expected_asset": img.get("expected_asset")
                or {"asset_type": "chain_bg", "variant_type": bg_id},
            "applies_to_shots": list(img.get("applies_to_shots", []) or []),
            "batch_index": bg_to_batch_index.get(bg_id, -1),
            "can_generate_after": can_generate_after,
            "api_call_shape": _api_call_shape_for(render_mode, has_ref),
        }

    # base prompt anchor diagnostic — structural only, no semantic analysis
    anchor_report = _build_base_prompt_anchor_report(gen_map, payloads)

    # prompt length advisory profile — no hard min, only max enforced separately
    prompt_lengths = [len(p["prompt_text"]) for p in payloads.values()]
    under_500 = sum(1 for length in prompt_lengths if length < W6_PROMPT_ADVISORY_MIN_LEN)
    prompt_length_profile = {
        "min": min(prompt_lengths) if prompt_lengths else 0,
        "max": max(prompt_lengths) if prompt_lengths else 0,
        "advisory_min_threshold": W6_PROMPT_ADVISORY_MIN_LEN,
        "under_500_count": under_500,
        "hard_max_threshold": W6_PROMPT_MAX_LEN,
    }

    unmapped = [
        # production fields W6 still does not emit (real API call territory)
        "image_bytes",            # actual PNG bytes
        "ref_file_path",          # PNG path on disk
        "sanitization_strategy",  # moderation retry sanitizer
        "validation_score",       # LVM image validation
        "validation_result",
        "openai_api_key",         # never mirrored
        "openai_client_handle",   # never mirrored
    ]

    counts = {
        "payloads": len(payloads),
        "groups": len(groups),
        "render_batches": len(batches),
        "shots_mapped": len(shot_map),
        "images_generate_method": sum(
            1 for p in payloads.values()
            if p["api_call_shape"]["client_method"] == "images.generate"
        ),
        "images_edit_method": sum(
            1 for p in payloads.values()
            if p["api_call_shape"]["client_method"] == "images.edit"
        ),
        "fallback_independent_no_base": sum(
            1 for p in payloads.values()
            if p["reference_policy"] == "fallback_independent_no_base"
        ),
    }

    return {
        "schema_version": 1,
        "stage": W6_STAGE,
        "plan_version": PLAN_VERSION,
        "image_generation_backend": W6_IMAGE_BACKEND,
        "payloads": payloads,
        "groups": groups,
        "render_batches": batches,
        "shot_to_background_image": shot_map,
        "counts": counts,
        "base_prompt_anchor_report": anchor_report,
        "prompt_length_profile": prompt_length_profile,
        "unmapped_to_production_fields": unmapped,
    }


def _build_base_prompt_anchor_report(gen_map: dict, payloads: dict) -> List[dict]:
    """Per-group structural anchor diagnostic. score-based, deterministic.

    Signals (each +1 unless noted):
    - +2: base_bg_id is None
    - +1: variant_count >= 3
    - +1: any member is weak_reference_variant
    - +1: max reference depth >= 2
    - +1: base prompt_brief length < 180
    - +1: base applies_to_shots count <= 1 AND variant_count >= 2

    score 0      -> low
    score 1..2   -> medium
    score >= 3   -> high

    Advisory only. Never blocks. No human-decision keys.
    """
    bg_images = gen_map.get("background_images", {}) or {}
    groups = gen_map.get("groups", []) or []
    report: List[dict] = []

    for g in groups:
        base_bg_id = g.get("base_bg_id")
        variant_bg_ids = list(g.get("variant_bg_ids") or [])
        member_bg_ids = list(g.get("member_bg_ids") or [])
        variant_count = len(variant_bg_ids)

        score = 0
        reasons: List[str] = []

        if not base_bg_id:
            score += 2
            reasons.append("no_base_bg_id")

        if variant_count >= 3:
            score += 1
            reasons.append("variant_count_ge_3")

        if any(
            bg_images.get(m, {}).get("render_mode") == "weak_reference_variant"
            for m in member_bg_ids
        ):
            score += 1
            reasons.append("group_includes_weak_reference_variant")

        # reference depth — each variant chain to base, count hops to root
        max_depth = 0
        for v in variant_bg_ids:
            depth = 0
            cur = v
            visited: List[str] = []
            while cur and cur not in visited and depth < 16:
                visited.append(cur)
                nxt = bg_images.get(cur, {}).get("primary_reference_bg_id")
                if not nxt:
                    break
                depth += 1
                cur = nxt
            if depth > max_depth:
                max_depth = depth
        if max_depth >= 2:
            score += 1
            reasons.append("max_reference_depth_ge_2")

        if base_bg_id:
            base_brief = (bg_images.get(base_bg_id, {}).get("prompt_brief") or "")
            if len(base_brief) < 180:
                score += 1
                reasons.append("base_prompt_brief_length_lt_180")
            base_shot_count = len(bg_images.get(base_bg_id, {}).get("applies_to_shots") or [])
            if base_shot_count <= 1 and variant_count >= 2:
                score += 1
                reasons.append("base_applies_to_shots_le_1_and_variant_count_ge_2")

        if score == 0:
            risk_level = "low"
        elif score <= 2:
            risk_level = "medium"
        else:
            risk_level = "high"

        # recommended_auto_fallback — string label, no human-decision enum
        if not base_bg_id:
            recommended_auto_fallback = "fallback_independent_no_base"
        elif any(
            bg_images.get(m, {}).get("render_mode") == "weak_reference_variant"
            for m in member_bg_ids
        ):
            recommended_auto_fallback = "use_weak_reference_continuity_for_variants"
        else:
            recommended_auto_fallback = (
                "generate_as_planned_then_fallback_to_independent_if_visual_validation_fails"
            )

        report.append({
            "group_id": g.get("group_id"),
            "base_bg_id": base_bg_id,
            "member_count": len(member_bg_ids),
            "variant_count": variant_count,
            "max_reference_depth": max_depth,
            "risk_score": score,
            "risk_level": risk_level,
            "risk_reasons": reasons,
            "recommended_auto_fallback": recommended_auto_fallback,
        })

    return report


def _build_image_payload_compatibility_report(
    *, plan: dict, gen_map: dict,
    production_diff_empty: bool, db_write_count: int,
    image_import_seen: bool, prev_run_id: str, missing_inputs: List[str],
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}
    payloads: Dict[str, dict] = plan.get("payloads", {}) or {}
    bg_images: Dict[str, dict] = gen_map.get("background_images", {}) or {}
    batches = plan.get("render_batches", []) or []
    shot_map = plan.get("shot_to_background_image", {}) or {}
    anchor_report = plan.get("base_prompt_anchor_report", []) or []

    inv["w5_inputs_present"] = {
        "pass": len(missing_inputs) == 0,
        "detail": {"missing": missing_inputs, "prev_run": prev_run_id},
    }

    expected = set(bg_images.keys())
    actual = set(payloads.keys())
    inv["all_payloads_covered"] = {
        "pass": bool(expected) and expected == actual,
        "detail": {
            "expected_count": len(expected),
            "actual_count": len(actual),
            "missing": sorted(expected - actual)[:10],
            "extra": sorted(actual - expected)[:10],
        },
    }

    backend = plan.get("image_generation_backend")
    wrong_model = [
        bg for bg, p in payloads.items()
        if p.get("image_model") != W6_IMAGE_BACKEND
        or p.get("api_call_shape", {}).get("model") != W6_IMAGE_BACKEND
    ]
    inv["payload_model_is_gpt_image_2"] = {
        "pass": backend == W6_IMAGE_BACKEND and len(wrong_model) == 0,
        "detail": {
            "plan_backend": backend,
            "violating_count": len(wrong_model),
            "violating_sample": wrong_model[:5],
        },
    }

    # api_call_shape mirrors render_mode policy
    shape_violations: List[str] = []
    for bg, p in payloads.items():
        method = p.get("api_call_shape", {}).get("client_method")
        has_ref = len(p.get("reference_images") or []) > 0
        if has_ref and method != "images.edit":
            shape_violations.append(f"{bg}:ref_present_but_method_{method}")
        if not has_ref and method != "images.generate":
            shape_violations.append(f"{bg}:no_ref_but_method_{method}")
    inv["api_call_shape_matches_render_mode"] = {
        "pass": len(shape_violations) == 0,
        "detail": {"violating": shape_violations[:5]},
    }

    # reference_payloads_point_to_existing_payloads
    ref_unknown: List[str] = []
    for bg, p in payloads.items():
        for ref in (p.get("reference_images") or []):
            rid = ref.get("bg_id")
            if rid not in payloads:
                ref_unknown.append(f"{bg}:{rid}")
    inv["reference_payloads_point_to_existing_payloads"] = {
        "pass": len(ref_unknown) == 0,
        "detail": {"violating": ref_unknown[:5]},
    }

    # references point to earlier batch
    bg_to_batch = {bg: i for i, batch in enumerate(batches) for bg in batch}
    bad_batch_ref: List[str] = []
    for bg, p in payloads.items():
        for ref in (p.get("reference_images") or []):
            rid = ref.get("bg_id")
            if rid not in bg_to_batch:
                bad_batch_ref.append(f"{bg}:ref_{rid}_unknown_batch")
                continue
            if bg_to_batch[rid] >= bg_to_batch.get(bg, 0):
                bad_batch_ref.append(
                    f"{bg}(batch={bg_to_batch.get(bg)})_refs_{rid}(batch={bg_to_batch[rid]})"
                )
    inv["references_point_to_earlier_batch"] = {
        "pass": len(bad_batch_ref) == 0,
        "detail": {"violating": bad_batch_ref[:5]},
    }

    # independent_payloads_have_no_refs
    independent_with_ref: List[str] = [
        bg for bg, p in payloads.items()
        if p.get("render_mode") == "independent_text_to_image"
        and len(p.get("reference_images") or []) != 0
    ]
    inv["independent_payloads_have_no_refs"] = {
        "pass": len(independent_with_ref) == 0,
        "detail": {"violating": independent_with_ref[:5]},
    }

    # referenced_payloads_have_refs — derive/weak modes with non-fallback policy
    referenced_missing_ref: List[str] = []
    for bg, p in payloads.items():
        if p.get("render_mode") in ("derive_from_base_reference", "weak_reference_variant"):
            if p.get("reference_policy") == "fallback_independent_no_base":
                continue  # mode declared but policy already fell back at W5
            if not (p.get("reference_images") or []):
                referenced_missing_ref.append(bg)
    inv["referenced_payloads_have_refs"] = {
        "pass": len(referenced_missing_ref) == 0,
        "detail": {"violating": referenced_missing_ref[:5]},
    }

    # shot_to_background_image carried exact — W5 shot_map == W6 shot_map
    w5_shot_map = gen_map.get("shot_to_background_image", {}) or {}
    inv["shot_to_background_image_coverage_carried_exact"] = {
        "pass": w5_shot_map == shot_map,
        "detail": {
            "w5_count": len(w5_shot_map),
            "w6_count": len(shot_map),
            "diff_sample": sorted(set(w5_shot_map.items()) ^ set(shot_map.items()))[:5],
        },
    }

    # payload_prompt_not_empty
    empty_prompts = [bg for bg, p in payloads.items() if not (p.get("prompt_text") or "").strip()]
    inv["payload_prompt_not_empty"] = {
        "pass": len(empty_prompts) == 0,
        "detail": {"violating": empty_prompts[:5]},
    }

    # payload_prompt_length_not_over_6000
    over_max = [
        f"{bg}:{len(p.get('prompt_text') or '')}"
        for bg, p in payloads.items()
        if len(p.get("prompt_text") or "") > W6_PROMPT_MAX_LEN
    ]
    inv["payload_prompt_length_not_over_6000"] = {
        "pass": len(over_max) == 0,
        "detail": {"violating": over_max[:5], "hard_max": W6_PROMPT_MAX_LEN},
    }

    # no_human_decision_field — scan payloads + anchor_report for banned keys
    banned_keys = {"needs_user_decision", "manual_review_required", "rollup",
                   "pending_human_decision", "human_review", "needs_approval"}
    bad_decision: List[str] = []
    for bg, p in payloads.items():
        if any(bk in p for bk in banned_keys):
            bad_decision.append(bg)
    for entry in anchor_report:
        if any(bk in entry for bk in banned_keys):
            bad_decision.append(entry.get("group_id", "?"))
    inv["no_human_decision_field"] = {
        "pass": len(bad_decision) == 0,
        "detail": {"violating": bad_decision[:5]},
    }

    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }
    inv["image_call_zero"] = {
        "pass": not image_import_seen,
        "detail": "no gemini_image_client / fal / openai images module import",
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


def _render_w6_html(run_meta: dict, plan: dict, report: dict, run_dir: Path) -> None:
    def esc(x):
        return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    counts = plan.get("counts", {}) or {}
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td class=\"{'pass' if v['pass'] else 'fail'}\">{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    anchor_rows = "".join(
        f"<tr><td>{esc(e.get('group_id'))}</td>"
        f"<td>{esc(e.get('base_bg_id'))}</td>"
        f"<td>{e.get('variant_count')}</td>"
        f"<td>{e.get('max_reference_depth')}</td>"
        f"<td>{e.get('risk_score')}</td>"
        f"<td class=\"{esc(e.get('risk_level'))}\">{esc(e.get('risk_level'))}</td>"
        f"<td>{esc(', '.join(e.get('risk_reasons') or []))}</td>"
        f"<td>{esc(e.get('recommended_auto_fallback'))}</td></tr>"
        for e in (plan.get("base_prompt_anchor_report") or [])
    )
    sample_payload_items = list((plan.get("payloads") or {}).items())[:2]
    sample_payloads = {k: v for k, v in sample_payload_items}
    prof = plan.get("prompt_length_profile", {}) or {}
    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice W6 {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em;}}
table{{border-collapse:collapse;margin-bottom:1em}} td,th{{border:1px solid #ccc;padding:4px 8px}}
.metric{{display:inline-block;margin:0 1em 1em 0;padding:1em;border:1px solid #ddd;border-radius:6px}}
.fail{{color:#b00}} .pass{{color:#080}} .low{{color:#080}} .medium{{color:#b87}} .high{{color:#b00}}</style></head>
<body>
<h1>background_pipeline_slice W6 — {esc(run_meta.get('run_id'))}</h1>
<p>stage: <b>{esc(run_meta.get('stage'))}</b> | plan_version: <b>{esc(run_meta.get('plan_version'))}</b>
| run_status: <b class=\"{'fail' if run_meta.get('run_status')!='succeeded' else 'pass'}\">{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}
| derived_from: {esc(run_meta.get('derived_from'))}
| image_backend: <b>{esc(run_meta.get('image_generation_backend'))}</b></p>
<div>
<div class=\"metric\"><div>payloads</div><b>{counts.get('payloads', 0)}</b></div>
<div class=\"metric\"><div>groups</div><b>{counts.get('groups', 0)}</b></div>
<div class=\"metric\"><div>render batches</div><b>{counts.get('render_batches', 0)}</b></div>
<div class=\"metric\"><div>shots mapped</div><b>{counts.get('shots_mapped', 0)}</b></div>
<div class=\"metric\"><div>images.generate</div><b>{counts.get('images_generate_method', 0)}</b></div>
<div class=\"metric\"><div>images.edit</div><b>{counts.get('images_edit_method', 0)}</b></div>
<div class=\"metric\"><div>fallback no_base</div><b>{counts.get('fallback_independent_no_base', 0)}</b></div>
<div class=\"metric\"><div>validation</div><b>{'PASS' if report and report.get('all_pass') else 'FAIL'}</b></div>
</div>
<h2>W6 Invariants</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Base prompt anchor report</h2>
<table>
<tr><th>group_id</th><th>base_bg_id</th><th>variants</th><th>max depth</th><th>score</th><th>risk</th><th>reasons</th><th>recommended_auto_fallback</th></tr>
{anchor_rows}
</table>
<h2>Prompt length profile (advisory)</h2>
<pre>{esc(json.dumps(prof, ensure_ascii=False, indent=2))}</pre>
<h2>Sample payloads (first 2)</h2><pre>{esc(json.dumps(sample_payloads, ensure_ascii=False, indent=2))}</pre>
<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
<details><summary>raw image_payload_plan (HTML preview only, slice 60000 chars; full JSON preserved)</summary><pre>{esc(json.dumps(plan, ensure_ascii=False, indent=2))[:60000]}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


def _w6_main(args, run_dir: Path, run_id: str) -> int:
    """W6 stage — derive image payload plan from a prior W5 success run."""
    global _DB_WRITE_COUNT
    prev_run_dir = Path(args.derive_image_payloads_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir
    artifacts = _load_w5_artifacts(prev_run_dir)
    missing = artifacts.get("_missing", [])
    outputs: List[str] = []
    failed: List[str] = []
    run_status = "succeeded"
    exit_code = 0

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W6_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model": None,
        "image_generation_backend": W6_IMAGE_BACKEND,
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed,
    }

    if missing:
        failed.append("w5_inputs_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    gen_map = artifacts["gen_map"]

    plan = _build_image_payload_plan(gen_map)
    (run_dir / "background_image_payload_plan.json").write_text(
        json.dumps(plan, ensure_ascii=False, indent=2)
    )
    outputs.append("background_image_payload_plan.json")

    report = _build_image_payload_compatibility_report(
        plan=plan, gen_map=gen_map,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_import_seen=_check_image_imports_present(),
        prev_run_id=prev_run_dir.name,
        missing_inputs=missing,
    )
    (run_dir / "image_payload_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("image_payload_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed:
            failed.append(name)
    if failed:
        run_status = "validation_failed"
        exit_code = 1

    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed
    run_meta["outputs"] = outputs
    _render_w6_html(run_meta, plan, report, run_dir)
    outputs.append("index.html")
    run_meta["outputs"] = outputs
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    _maybe_print_imports(args)
    return exit_code


# ─────────────────────────────────────────────────────────────────────────────
# W7 — plate prompt plan (LLM rewrite W6 payload prompts into background plates)
# ─────────────────────────────────────────────────────────────────────────────

W7_STAGE = "w7_plate_prompt_plan"
W7_PROMPT_OVER_2000_ADVISORY = 2000
W7_PROMPT_HARD_MAX = W6_PROMPT_MAX_LEN  # 6000, shared


def _load_w6_artifacts(prev_run_dir: Path) -> dict:
    """W7 input — load W6 success run artifacts read-only."""
    required = {
        "plan": "background_image_payload_plan.json",
        "report": "image_payload_compatibility_report.json",
        "run_meta": "run_meta.json",
    }
    out: Dict[str, Any] = {}
    missing: List[str] = []
    for key, fname in required.items():
        p = prev_run_dir / fname
        if not p.exists():
            missing.append(fname)
            continue
        out[key] = json.loads(p.read_text())
    out["_missing"] = missing
    out["_prev_run_id"] = prev_run_dir.name
    return out


def _classify_prompt_role(render_mode: str, reference_policy: str) -> str:
    if reference_policy == "fallback_independent_no_base":
        return "fallback_independent_plate"
    if render_mode == "independent_text_to_image":
        return "base_plate"
    if render_mode == "derive_from_base_reference":
        return "strict_variant_delta"
    if render_mode == "weak_reference_variant":
        return "weak_variant_delta"
    return "base_plate"


def _classify_base_prompt_strategy(risk_level: str) -> str:
    """W7b — per-group strategy for how the base anchors variants.

    Narrow rule: high-risk groups force the runtime to broaden the base
    anchor and treat variants as weak-continuity deltas instead of strict
    geometry matches. low/medium keep standard policy.
    """
    if (risk_level or "").lower() == "high":
        return "weak_reference_variants"
    return "standard"


def _classify_reference_strength(render_mode: str, group_risk: str) -> str:
    """W7b — per-payload reference interpretation strength.

    Default mapping (deterministic):
      - independent_text_to_image -> none
      - weak_reference_variant   -> weak
      - derive_from_base_reference -> strict
    Override: if the group's risk_level is high, derive-from-base variants
    are downgraded to weak so the runtime applies spatial-tone continuity
    instead of strict geometry. The API method stays images.edit; only the
    prompt/reference interpretation changes.
    """
    if render_mode == "independent_text_to_image":
        return "none"
    if render_mode == "weak_reference_variant":
        return "weak"
    if render_mode == "derive_from_base_reference":
        if (group_risk or "").lower() == "high":
            return "weak"
        return "strict"
    return "none"


def _build_bg_to_group_risk(w6_plan: dict) -> Dict[str, str]:
    """Map every bg_id to its group's risk_level (from W6 anchor report)."""
    anchor = {e["group_id"]: e for e in (w6_plan.get("base_prompt_anchor_report") or [])}
    groups = w6_plan.get("groups", []) or []
    bg_to_risk: Dict[str, str] = {}
    for g in groups:
        gid = g.get("group_id")
        risk = (anchor.get(gid, {}) or {}).get("risk_level") or "low"
        for m in (g.get("member_bg_ids") or []):
            bg_to_risk[m] = risk
    return bg_to_risk


# ─────────────────────────────────────────────────────────────────────────────
# W7c — source context resolution & per-group context builder
# ─────────────────────────────────────────────────────────────────────────────

W7C_CHAIN_MAX_HOPS = 8  # bounded; W6 -> W5 -> W4 -> W3 == 3 hops in practice


def _resolve_source_bundle_via_derived_from_chain(start_run_dir: Path,
                                                  max_hops: int = W7C_CHAIN_MAX_HOPS) -> dict:
    """Walk the derived_from chain (W6 -> W5 -> W4 -> W3) until a run dir
    that contains source_bundle.json is found. Bounded hops; fail-closed.

    Returns {"source_bundle": dict, "chain": [run_id, ...], "root_run_id": str}.
    Raises FileNotFoundError if the chain breaks or exhausts hops without a
    source_bundle.json.
    """
    parent_dir = start_run_dir.parent
    chain: List[str] = []
    cur_dir = start_run_dir
    for hop in range(max_hops + 1):
        rm_path = cur_dir / "run_meta.json"
        if not rm_path.exists():
            raise FileNotFoundError(
                f"derived_from chain broken at hop={hop}: {rm_path} missing"
            )
        chain.append(cur_dir.name)
        sb_path = cur_dir / "source_bundle.json"
        if sb_path.exists():
            bundle = json.loads(sb_path.read_text())
            return {"source_bundle": bundle, "chain": chain, "root_run_id": cur_dir.name}
        rm = json.loads(rm_path.read_text())
        prev_id = rm.get("derived_from")
        if not prev_id:
            raise FileNotFoundError(
                f"derived_from chain terminated at {cur_dir.name} without "
                f"source_bundle.json (root reached, hop={hop})"
            )
        nxt = parent_dir / prev_id
        if not nxt.exists():
            raise FileNotFoundError(
                f"derived_from points to non-existent dir: {prev_id} (from {cur_dir.name})"
            )
        cur_dir = nxt
    raise FileNotFoundError(
        f"derived_from chain exceeded max_hops={max_hops} without source_bundle.json; "
        f"visited={chain}"
    )


def _safe_parse_json_field(value: Any) -> Any:
    """selected_shots.t2i_variations_json / visible_entities_json may be stored
    as raw JSON-text strings or already as objects. Parse defensively; if it
    fails, return None (caller decides how to carry)."""
    if value is None:
        return None
    if isinstance(value, (dict, list)):
        return value
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return None
        try:
            return json.loads(s)
        except Exception:
            return None
    return None


def _extract_t2i_anchors(t2i_variations: Any) -> List[Dict[str, Any]]:
    """Per-variant carry of source_facts / visual_inferences / owned_object_usage
    plus variant_label. Pure structural carry; no semantic filtering.

    Additionally surfaces `anchor_objects`: a deterministic subset of
    owned_object_usage entries whose usage_kind enum value equals 'anchor'.
    This is an enum-only filter (no word matching); the LLM downstream still
    decides whether each token is background-owned or character-owned.
    """
    if not isinstance(t2i_variations, list):
        return []
    out: List[Dict[str, Any]] = []
    for v in t2i_variations:
        if not isinstance(v, dict):
            continue
        oou = v.get("owned_object_usage")
        anchor_objects: List[Dict[str, Any]] = []
        if isinstance(oou, list):
            for e in oou:
                if not isinstance(e, dict):
                    continue
                if e.get("usage_kind") == "anchor":
                    anchor_objects.append({
                        "owned_token": e.get("owned_token"),
                        "source_phrase": e.get("source_phrase"),
                    })
        out.append({
            "variant_label": v.get("variant_label"),
            "source_facts": v.get("source_facts"),
            "visual_inferences": v.get("visual_inferences"),
            "owned_object_usage": oou,
            "anchor_objects": anchor_objects,
            "applied_frame_spatial_constraint_ids": v.get("applied_frame_spatial_constraint_ids"),
        })
    return out


def _extract_visible_entities_ids(ve_parsed: Any) -> List[str]:
    """Pull short_id-like strings from visible_entities (helper only, no semantic).
    visible_entities may be a list of dicts each with id/short_id, or other
    shapes; just collect any short_id-ish string field."""
    if isinstance(ve_parsed, list):
        ids: List[str] = []
        for e in ve_parsed:
            if isinstance(e, dict):
                for k in ("short_id", "id", "entity_short_id"):
                    val = e.get(k)
                    if isinstance(val, str) and val:
                        ids.append(val)
                        break
        return ids
    if isinstance(ve_parsed, dict):
        out: List[str] = []
        for k in ("characters", "props", "backgrounds", "items"):
            sub = ve_parsed.get(k)
            if isinstance(sub, list):
                for e in sub:
                    if isinstance(e, dict):
                        for kk in ("short_id", "id", "entity_short_id"):
                            val = e.get(kk)
                            if isinstance(val, str) and val:
                                out.append(val)
                                break
        return out
    return []


def _build_per_group_source_context(w6_plan: dict, source_bundle: dict) -> Dict[str, dict]:
    """For every W6 group, gather exact-id source context: location record +
    selected_shot raw fields filtered by the group's shot_keys. No semantic
    extraction — only id matching + verbatim carry."""
    groups = w6_plan.get("groups", []) or []
    loc_index = {l.get("loc_id"): l for l in (source_bundle.get("locations_for_llm") or [])}
    shots_index = {s.get("shot_key"): s for s in (source_bundle.get("selected_shots") or [])}

    out: Dict[str, dict] = {}
    for g in groups:
        gid = g.get("group_id")
        loc_id = g.get("loc_id")
        shot_keys = list(g.get("shot_keys") or [])
        loc_record = loc_index.get(loc_id)
        loc_view = None
        if loc_record:
            loc_view = {
                "loc_id": loc_record.get("loc_id"),
                "name": loc_record.get("name"),
                "description": loc_record.get("description"),
                "space_profile": loc_record.get("space_profile"),
            }
        shot_views: List[Dict[str, Any]] = []
        for sk in shot_keys:
            s = shots_index.get(sk)
            if not s:
                continue
            t2i_parsed = _safe_parse_json_field(s.get("t2i_variations_json"))
            ve_parsed = _safe_parse_json_field(s.get("visible_entities_json"))
            shot_views.append({
                "shot_key": s.get("shot_key"),
                "scene_index": s.get("scene_index"),
                "shot_index": s.get("shot_index"),
                "scene_type": s.get("scene_type"),
                "screenplay_scene_heading": s.get("screenplay_scene_heading"),
                "beat_title": s.get("beat_title"),
                "scene_summary": s.get("scene_summary"),
                "shot_description": s.get("shot_description"),
                "still_frame_prompt": s.get("still_frame_prompt"),
                "t2i_variant_anchors": _extract_t2i_anchors(t2i_parsed),
                "visible_entity_short_ids": _extract_visible_entities_ids(ve_parsed),
            })
        out[gid] = {
            "group_id": gid,
            "loc_id": loc_id,
            "shot_keys": shot_keys,
            "location": loc_view,
            "shots": shot_views,
        }
    return out


LLM_PLATE_PROMPT_SCHEMA = {
    "type": "object",
    "required": ["plate_prompts", "base_plate_group_guides"],
    "properties": {
        "plate_prompts": {
            "type": "object",
            "additionalProperties": {
                "type": "object",
                "required": ["plate_prompt_text", "consistency_intent", "variant_delta"],
                "properties": {
                    "plate_prompt_text": {"type": "string"},
                    "consistency_intent": {"type": "string"},
                    "variant_delta": {"type": "string"},
                },
            },
        },
        "base_plate_group_guides": {
            "type": "object",
            "additionalProperties": {
                "type": "object",
                "required": [
                    "shared_plate_intent", "shared_continuity_constraints",
                    "variant_delta_policy", "auto_strategy",
                    # W7c — source-derived constraints, generic field names. LLM picks
                    # the actual anchor words by reading the source context input.
                    "source_context_summary",
                    "scale_and_default_avoidance",
                    "must_preserve_spatial_anchors",
                    "state_transition_anchors",
                ],
                "properties": {
                    "shared_plate_intent": {"type": "string"},
                    "shared_continuity_constraints": {"type": "string"},
                    "variant_delta_policy": {"type": "string"},
                    "auto_strategy": {"type": "string"},
                    "source_context_summary": {"type": "string"},
                    "scale_and_default_avoidance": {"type": "string"},
                    "must_preserve_spatial_anchors": {"type": "string"},
                    "state_transition_anchors": {"type": "string"},
                },
            },
        },
    },
}


def _build_placeholder_plate_plan(w6_plan: dict) -> dict:
    """Dry-run — placeholder plate_prompt_text carries W6 prompt_text verbatim. status=placeholder_dry_run."""
    payloads = w6_plan.get("payloads", {}) or {}
    groups = w6_plan.get("groups", []) or []
    anchor = {e["group_id"]: e for e in (w6_plan.get("base_prompt_anchor_report") or [])}
    bg_to_risk = _build_bg_to_group_risk(w6_plan)

    plate_prompts: Dict[str, dict] = {}
    for bg_id, p in payloads.items():
        role = _classify_prompt_role(p.get("render_mode", ""), p.get("reference_policy", ""))
        group_risk = bg_to_risk.get(bg_id, "low")
        plate_prompts[bg_id] = {
            "bg_id": bg_id,
            "image_title": p.get("image_title"),
            "image_model": W6_IMAGE_BACKEND,
            "render_mode": p.get("render_mode"),
            "api_call_shape": dict(p.get("api_call_shape") or {}),
            "reference_images": list(p.get("reference_images") or []),
            "expected_asset": dict(p.get("expected_asset") or {}),
            "applies_to_shots": list(p.get("applies_to_shots") or []),
            "original_prompt_text": p.get("prompt_text", ""),
            "plate_prompt_text": p.get("prompt_text", ""),
            "prompt_role": role,
            "reference_strength": _classify_reference_strength(p.get("render_mode", ""), group_risk),
            "group_risk_level": group_risk,
            "consistency_intent": "",
            "variant_delta": "",
            "prompt_status": "placeholder_dry_run",
        }

    group_guides: Dict[str, dict] = {}
    for g in groups:
        gid = g.get("group_id")
        risk = (anchor.get(gid, {}) or {}).get("risk_level") or "low"
        auto = (anchor.get(gid, {}) or {}).get("recommended_auto_fallback") or ""
        group_guides[gid] = {
            "group_id": gid,
            "base_bg_id": g.get("base_bg_id"),
            "shared_plate_intent": "",
            "shared_continuity_constraints": "",
            "variant_delta_policy": "",
            "risk_level_from_w6": risk,
            "auto_strategy": auto,
            "base_prompt_strategy": _classify_base_prompt_strategy(risk),
            # W7c — source-derived fields populated only by LLM generate path.
            "source_context_summary": "",
            "scale_and_default_avoidance": "",
            "must_preserve_spatial_anchors": "",
            "state_transition_anchors": "",
            "guide_status": "placeholder_dry_run",
        }

    return _assemble_plate_plan(w6_plan, plate_prompts, group_guides,
                                model_used=None,
                                generation_status="placeholder_dry_run")


def _generate_llm_plate_plan(w6_plan: dict, *, model: str,
                             source_context: Optional[Dict[str, dict]] = None) -> dict:
    """--generate — LLM rewrites payload prompts into background plate prompts.

    source_context (W7c): per-group dict from _build_per_group_source_context.
    When present, it is passed to the LLM so each group_guide can include
    source-derived constraints (scale/locality/default-avoidance/spatial
    anchors/state transition anchors). The runtime does NOT pick anchor words;
    the LLM reads source and decides.
    """
    import litellm  # lazy import — dry-run path must not import it

    payloads = w6_plan.get("payloads", {}) or {}
    groups = w6_plan.get("groups", []) or []
    anchor = {e["group_id"]: e for e in (w6_plan.get("base_prompt_anchor_report") or [])}
    bg_to_risk = _build_bg_to_group_risk(w6_plan)

    # W10 — per-bg recurring anchor token aggregate.
    # source_context (when provided) holds the bg's shots[].t2i_variant_anchors[]
    # .anchor_objects (already filtered by usage_kind=='anchor'). Group those
    # tokens across the bg's shots and count occurrences. Tokens that appear in
    # at least two shots are surfaced as recurring_anchor_candidates so the LLM
    # does not omit them. Token identity equality only — no word matching, no
    # interpretation of what each token means.
    def _aggregate_recurring_anchors(group_ctx: Optional[dict],
                                     applies_to_shots: Optional[List[str]]) -> List[Dict[str, Any]]:
        if not isinstance(group_ctx, dict):
            return []
        target_keys = list(applies_to_shots or [])
        if not target_keys:
            return []
        per_token: Dict[str, Dict[str, Any]] = {}
        per_token_shots: Dict[str, List[str]] = {}
        for shot in (group_ctx.get("shots") or []):
            shot_key = shot.get("shot_key")
            if shot_key not in target_keys:
                continue  # only count shots that actually belong to this bg
            anchors_seen_in_shot: List[str] = []
            for variant in (shot.get("t2i_variant_anchors") or []):
                for a in (variant.get("anchor_objects") or []):
                    tok = a.get("owned_token")
                    if not tok or tok in anchors_seen_in_shot:
                        continue
                    anchors_seen_in_shot.append(tok)
                    entry = per_token.setdefault(tok, {
                        "owned_token": tok, "occurrences": 0, "source_phrases": [],
                    })
                    entry["occurrences"] += 1
                    phrase = a.get("source_phrase") or ""
                    if phrase and phrase not in entry["source_phrases"]:
                        entry["source_phrases"].append(phrase)
                    shot_list = per_token_shots.setdefault(tok, [])
                    if shot_key and shot_key not in shot_list:
                        shot_list.append(shot_key)
        # If this bg only has 1 applies_to_shot, "recurring" reduces to all
        # anchor tokens in that single shot (occurrences>=1). Otherwise we
        # require the token to appear in >=2 distinct applies_to_shots.
        min_shot_count = 1 if len(target_keys) <= 1 else 2
        out: List[Dict[str, Any]] = []
        for tok, entry in per_token.items():
            shot_count = len(per_token_shots.get(tok, []))
            entry["distinct_shot_count"] = shot_count
            if shot_count >= min_shot_count:
                out.append(entry)
        out.sort(key=lambda e: (-e["distinct_shot_count"], e["owned_token"]))
        return out

    sc_pre = source_context or {}
    # bg_id -> group_id from W6 groups
    bg_to_group_id: Dict[str, str] = {}
    for g in groups:
        gid = g.get("group_id")
        for m in (g.get("member_bg_ids") or []):
            bg_to_group_id[m] = gid

    compact_payloads = {
        bg_id: {
            "bg_id": bg_id,
            "image_title": p.get("image_title"),
            "render_mode": p.get("render_mode"),
            "reference_policy": p.get("reference_policy"),
            "primary_reference_bg_id": (
                (p.get("reference_images") or [{}])[0].get("bg_id")
                if p.get("reference_images") else None
            ),
            "original_prompt_text": p.get("prompt_text", ""),
            "applies_to_shots": p.get("applies_to_shots") or [],
            "batch_index": p.get("batch_index", -1),
            # W7d — make the runtime reference-strength decision visible to the LLM
            # so per-payload wording matches the chosen reference policy.
            "group_risk_level": bg_to_risk.get(bg_id, "low"),
            "reference_strength": _classify_reference_strength(
                p.get("render_mode", ""), bg_to_risk.get(bg_id, "low")
            ),
            # W10 — recurring anchor candidates aggregated from the bg's own
            # applies_to_shots only (filtered out of the group's full shot list).
            "recurring_anchor_candidates": _aggregate_recurring_anchors(
                sc_pre.get(bg_to_group_id.get(bg_id)),
                p.get("applies_to_shots") or [],
            ),
        }
        for bg_id, p in payloads.items()
    }
    sc = sc_pre
    compact_groups = [
        {
            "group_id": g.get("group_id"),
            "base_bg_id": g.get("base_bg_id"),
            "member_bg_ids": g.get("member_bg_ids", []),
            "variant_bg_ids": g.get("variant_bg_ids", []),
            "risk_level_from_w6": (anchor.get(g.get("group_id"), {}) or {}).get("risk_level"),
            "risk_reasons_from_w6": (anchor.get(g.get("group_id"), {}) or {}).get("risk_reasons", []),
            "base_prompt_strategy": _classify_base_prompt_strategy(
                (anchor.get(g.get("group_id"), {}) or {}).get("risk_level", "low")
            ),
            # W7c — per-group exact-id source context (location + group shots only).
            "source_context": sc.get(g.get("group_id")),
        }
        for g in groups
    ]

    system_prompt = (
        "You rewrite per-image prompts into stable background plate prompts for a film "
        "scene image pipeline. Output JSON only, matching the provided schema. "
        "For each bg_id, produce plate_prompt_text that describes the spatial plate of the "
        "background as a stable shared layout (architecture, scale, materials, lighting "
        "baseline) rather than a per-shot camera framing. Avoid character-action or "
        "camera-movement wording unless strictly required to anchor the space. For base "
        "plates (independent or fallback) write the prompt as the shared anchor for all "
        "variants in its group. For variants, the prompt should articulate the delta from "
        "the referenced base (state/time/special cue) while assuming spatial continuity. "
        "Do NOT pad prompts to reach any minimum length. Do NOT truncate prompts. Each "
        "plate_prompt_text must remain at most 6000 characters. Approximately 300-2000 "
        "characters is a natural range but is not enforced. "
        "Also emit base_plate_group_guides keyed by group_id describing the shared plate "
        "intent, continuity constraints, variant delta policy, and an auto_strategy label. "
        "Do not introduce sample-specific location labels, named characters, episode "
        "identifiers, or any human-decision/approval/manual_review fields. "
        "consistency_intent and variant_delta must be strings (short sentences or "
        "comma-separated phrases), not arrays. Use 'base_plate' role naturally for "
        "independents and 'strict_variant_delta'/'weak_variant_delta'/"
        "'fallback_independent_plate' as appropriate, although your output only emits "
        "plate_prompt_text/consistency_intent/variant_delta for each bg_id (the role is "
        "assigned by the runtime). "
        # W7b — high-risk group reference policy.
        "Each group input includes a 'base_prompt_strategy' field. When base_prompt_strategy "
        "is 'weak_reference_variants' (signaled by high risk in the W6 anchor report, e.g. "
        "many variants, deep reference chains, or weak-reference members in the group), "
        "treat the base plate as a BROADER group anchor: integrate the shared anchors "
        "implied by the group's member set (shared_plate_intent and shared_continuity_"
        "constraints in your group_guide) so that variants can be derived without strict "
        "geometry matching. For variants in such groups, write the plate prompt assuming "
        "the reference will be used for spatial tone and layout continuity only, NOT for "
        "strict geometry matching; do not demand pixel-level alignment with the base. For "
        "groups with base_prompt_strategy='standard', keep the standard behavior (variants "
        "may rely on stricter geometric continuity with the base). "
        # W7c — source-derived constraints. The runtime never picks anchor words.
        "Each group input also includes a 'source_context' object with the upstream "
        "source bundle's location record and the group's selected_shots (raw fields: "
        "screenplay headings, beats, scene summaries, shot descriptions, still frame "
        "prompts, t2i variant anchors). USE this source context to derive: "
        "(1) source_context_summary — one or two sentences summarising what the source "
        "says about this location and the group's shot situations; "
        "(2) scale_and_default_avoidance — scale/locality/age/material constraints derived "
        "from the source; explicitly call out what generic defaults the image should NOT "
        "drift toward (do not invent defaults the source does not support); "
        "(3) must_preserve_spatial_anchors — the spatial elements the source repeatedly "
        "implies the variants share (architectural openings, surfaces, structural elements, "
        "etc.); "
        "(4) state_transition_anchors — the logical anchors that link different states "
        "across variants (state-change cues, continuity markers, presence/absence logic). "
        "These four fields are mandatory strings in every group_guide. "
        "Reflect the derived constraints back into plate_prompt_text/consistency_intent/"
        "variant_delta — do not write generic 'modern minimalist' or 'luxury' descriptions "
        "when the source does not support them. Stay faithful to what the source says, in "
        "whatever language the source uses. Do not import named characters, episode "
        "identifiers, or any sample-specific labels into the methodology fields; copy "
        "source-grounded descriptions only when they come from the provided source_context. "
        # W7c/W7e — ownership boundary (abstract category names only; do not name
        # concrete entity instances or sample-specific furnishings in the methodology).
        "Ownership boundary: character/entity instances and character-state variants are "
        "NOT background-owned anchors. If the source_context mentions them, use only the "
        "surrounding background/environmental evidence they imply, such as surface "
        "condition, marks or stains on background surfaces, architectural openings, "
        "persistent fixtures and furnishings when source-supported, lighting changes, and "
        "clean or absence states of the space. Leave the entity instance itself to the "
        "entity/character pipelines. Do not place character entities or character-state "
        "instances into must_preserve_spatial_anchors or any other background spatial "
        "anchor field. "
        # W10 — per-bg recurring background-owned anchor coverage.
        "Per-bg anchor coverage: every payload entry in the input includes a "
        "'recurring_anchor_candidates' list aggregated from the bg's "
        "applies_to_shots (anchor_objects with usage_kind='anchor' that appear in at "
        "least two distinct shots of that bg). Treat every recurring_anchor_candidate "
        "as a mandatory inclusion in plate_prompt_text UNLESS the token is clearly a "
        "character/entity instance (people, body parts, named characters) — in which "
        "case skip it. Use the source_phrases provided to phrase the anchor naturally; "
        "do not invent additional anchors that are not in any anchor_objects list of "
        "the bg's source_context. The recurring_anchor_candidates are background-"
        "owned by aggregation policy; if a token sounds character-owned, prefer to "
        "leave it out rather than include it. Anchors that only appear once (not in "
        "recurring_anchor_candidates) are optional and lower priority. "
        # W7d — per-bg source anchor reflection + reference-strength wording.
        "Do not leave the important source-derived anchors only in group_guide. For each "
        "plate_prompt_text, include the source-derived anchors relevant to that bg_id's "
        "own applies_to_shots and original_prompt_text whenever they affect background "
        "continuity (architectural openings, surfaces, lighting baseline, persistent "
        "spatial elements, applicable environmental state cues). The bg-level plate_"
        "prompt_text is what downstream image generation will actually consume, so it "
        "must independently carry the anchors that bind this bg to its group, instead "
        "of relying on the group_guide alone. "
        "Reference-strength wording: each payload input carries a 'reference_strength' "
        "field (none/strict/weak) and a 'group_risk_level' field. When reference_strength "
        "is 'weak', the plate_prompt_text, consistency_intent, and variant_delta must NOT "
        "use phrases like 'strict spatial continuity', 'strict geometry', 'pixel-perfect "
        "alignment', or 'exact match with reference'. Use weaker continuity language "
        "instead, such as 'shared spatial tone', 'shared layout continuity', 'loose "
        "structural alignment', or 'shared atmospheric continuity'. When reference_"
        "strength is 'strict', stricter continuity wording is allowed but is not "
        "mandatory. When reference_strength is 'none', do not assume any reference at all."
    )
    user_prompt = json.dumps(
        {
            "payloads": compact_payloads,
            "groups": compact_groups,
            "json_schema": LLM_PLATE_PROMPT_SCHEMA,
        },
        ensure_ascii=False,
    )
    routed = (
        model
        if model.startswith("gemini/") or not model.lower().startswith("gemini")
        else f"gemini/{model}"
    )
    if not (os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")):
        raise RuntimeError("missing GEMINI_API_KEY / GOOGLE_API_KEY env var")
    resp = litellm.completion(
        model=routed,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        response_format={"type": "json_object"},
    )
    raw_text = resp.choices[0].message.content
    decoder = json.JSONDecoder()
    stripped = raw_text.lstrip()
    parsed, end = decoder.raw_decode(stripped)
    trailing_ignored = bool(stripped[end:].strip())

    # schema validate
    import jsonschema  # type: ignore
    jsonschema.validate(parsed, LLM_PLATE_PROMPT_SCHEMA)

    return _merge_llm_plate_output(w6_plan, parsed, model_used=model,
                                   trailing_ignored=trailing_ignored)


def _merge_llm_plate_output(w6_plan: dict, llm_out: dict, *,
                            model_used: str, trailing_ignored: bool) -> dict:
    """Merge LLM-emitted plate fields with W6 carry. Strict bg/group coverage check."""
    payloads = w6_plan.get("payloads", {}) or {}
    groups = w6_plan.get("groups", []) or []
    anchor = {e["group_id"]: e for e in (w6_plan.get("base_prompt_anchor_report") or [])}

    llm_plates = llm_out.get("plate_prompts", {}) or {}
    llm_guides = llm_out.get("base_plate_group_guides", {}) or {}

    expected_bg = set(payloads.keys())
    actual_bg = set(llm_plates.keys())
    expected_groups = {g.get("group_id") for g in groups}
    actual_groups = set(llm_guides.keys())

    bg_missing = sorted(expected_bg - actual_bg)
    bg_extra = sorted(actual_bg - expected_bg)
    group_missing = sorted(expected_groups - actual_groups)
    group_extra = sorted(actual_groups - expected_groups)

    validation_errors: List[str] = []
    if bg_missing:
        validation_errors.append(f"plate_prompts missing bg_ids: {bg_missing[:5]}")
    if bg_extra:
        validation_errors.append(f"plate_prompts extra bg_ids: {bg_extra[:5]}")
    if group_missing:
        validation_errors.append(f"group_guides missing group_ids: {group_missing[:5]}")
    if group_extra:
        validation_errors.append(f"group_guides extra group_ids: {group_extra[:5]}")
    # per-bg sanity
    for bg_id, entry in llm_plates.items():
        if not (entry.get("plate_prompt_text") or "").strip():
            validation_errors.append(f"{bg_id}:plate_prompt_text_empty")
        elif len(entry["plate_prompt_text"]) > W7_PROMPT_HARD_MAX:
            validation_errors.append(
                f"{bg_id}:plate_prompt_text_over_{W7_PROMPT_HARD_MAX}"
            )

    bg_to_risk = _build_bg_to_group_risk(w6_plan)

    plate_prompts: Dict[str, dict] = {}
    for bg_id, p in payloads.items():
        role = _classify_prompt_role(p.get("render_mode", ""), p.get("reference_policy", ""))
        llm_entry = llm_plates.get(bg_id, {})
        group_risk = bg_to_risk.get(bg_id, "low")
        plate_prompts[bg_id] = {
            "bg_id": bg_id,
            "image_title": p.get("image_title"),
            "image_model": W6_IMAGE_BACKEND,
            "render_mode": p.get("render_mode"),
            "api_call_shape": dict(p.get("api_call_shape") or {}),
            "reference_images": list(p.get("reference_images") or []),
            "expected_asset": dict(p.get("expected_asset") or {}),
            "applies_to_shots": list(p.get("applies_to_shots") or []),
            "original_prompt_text": p.get("prompt_text", ""),
            "plate_prompt_text": llm_entry.get("plate_prompt_text", ""),
            "prompt_role": role,
            "reference_strength": _classify_reference_strength(p.get("render_mode", ""), group_risk),
            "group_risk_level": group_risk,
            "consistency_intent": llm_entry.get("consistency_intent", ""),
            "variant_delta": llm_entry.get("variant_delta", ""),
            "prompt_status": "generated" if bg_id in llm_plates else "validation_failed",
        }

    group_guides: Dict[str, dict] = {}
    for g in groups:
        gid = g.get("group_id")
        risk = (anchor.get(gid, {}) or {}).get("risk_level") or "low"
        auto_default = (anchor.get(gid, {}) or {}).get("recommended_auto_fallback") or ""
        llm_g = llm_guides.get(gid, {})
        group_guides[gid] = {
            "group_id": gid,
            "base_bg_id": g.get("base_bg_id"),
            "shared_plate_intent": llm_g.get("shared_plate_intent", ""),
            "shared_continuity_constraints": llm_g.get("shared_continuity_constraints", ""),
            "variant_delta_policy": llm_g.get("variant_delta_policy", ""),
            "risk_level_from_w6": risk,
            "auto_strategy": llm_g.get("auto_strategy") or auto_default,
            "base_prompt_strategy": _classify_base_prompt_strategy(risk),
            # W7c — source-derived constraints carried verbatim from LLM (no
            # runtime selection of anchor words; the LLM reads source_context).
            "source_context_summary": llm_g.get("source_context_summary", ""),
            "scale_and_default_avoidance": llm_g.get("scale_and_default_avoidance", ""),
            "must_preserve_spatial_anchors": llm_g.get("must_preserve_spatial_anchors", ""),
            "state_transition_anchors": llm_g.get("state_transition_anchors", ""),
            "guide_status": "generated" if gid in llm_guides else "validation_failed",
        }

    status = "generated" if not validation_errors else "validation_failed"
    return _assemble_plate_plan(
        w6_plan, plate_prompts, group_guides,
        model_used=model_used,
        generation_status=status,
        validation_errors=validation_errors,
        trailing_ignored=trailing_ignored,
    )


def _assemble_plate_plan(w6_plan: dict, plate_prompts: dict, group_guides: dict,
                         *, model_used, generation_status: str,
                         validation_errors: Optional[List[str]] = None,
                         trailing_ignored: bool = False) -> dict:
    """Common assembly — carries W6 fields and computes prompt_length_profile."""
    prompts_text = [
        (p.get("plate_prompt_text") or "") for p in plate_prompts.values()
    ]
    lengths = [len(t) for t in prompts_text]
    profile = {
        "min": min(lengths) if lengths else 0,
        "max": max(lengths) if lengths else 0,
        "over_2000_count": sum(1 for length in lengths if length > W7_PROMPT_OVER_2000_ADVISORY),
        "over_6000_count": sum(1 for length in lengths if length > W7_PROMPT_HARD_MAX),
        "advisory_over_threshold": W7_PROMPT_OVER_2000_ADVISORY,
        "hard_max_threshold": W7_PROMPT_HARD_MAX,
    }

    counts = {
        "plate_prompts": len(plate_prompts),
        "groups": len(group_guides),
        "role_base_plate": sum(1 for p in plate_prompts.values()
                               if p["prompt_role"] == "base_plate"),
        "role_strict_variant_delta": sum(1 for p in plate_prompts.values()
                                         if p["prompt_role"] == "strict_variant_delta"),
        "role_weak_variant_delta": sum(1 for p in plate_prompts.values()
                                       if p["prompt_role"] == "weak_variant_delta"),
        "role_fallback_independent_plate": sum(
            1 for p in plate_prompts.values()
            if p["prompt_role"] == "fallback_independent_plate"
        ),
        "reference_strength_none": sum(1 for p in plate_prompts.values()
                                       if p.get("reference_strength") == "none"),
        "reference_strength_strict": sum(1 for p in plate_prompts.values()
                                         if p.get("reference_strength") == "strict"),
        "reference_strength_weak": sum(1 for p in plate_prompts.values()
                                       if p.get("reference_strength") == "weak"),
        "groups_strategy_standard": sum(1 for g in group_guides.values()
                                        if g.get("base_prompt_strategy") == "standard"),
        "groups_strategy_weak_reference_variants": sum(
            1 for g in group_guides.values()
            if g.get("base_prompt_strategy") == "weak_reference_variants"
        ),
        "status_generated": sum(1 for p in plate_prompts.values()
                                if p["prompt_status"] == "generated"),
        "status_placeholder_dry_run": sum(1 for p in plate_prompts.values()
                                          if p["prompt_status"] == "placeholder_dry_run"),
        "status_validation_failed": sum(1 for p in plate_prompts.values()
                                        if p["prompt_status"] == "validation_failed"),
    }

    return {
        "schema_version": 1,
        "stage": W7_STAGE,
        "plan_version": PLAN_VERSION,
        "image_generation_backend": W6_IMAGE_BACKEND,
        "model_used": model_used,
        "generation_status": generation_status,
        "validation_errors": validation_errors or [],
        "raw_text_trailing_ignored": trailing_ignored,
        "plate_prompts": plate_prompts,
        "base_plate_group_guides": group_guides,
        "groups": w6_plan.get("groups", []),
        "render_batches": w6_plan.get("render_batches", []),
        "shot_to_background_image": w6_plan.get("shot_to_background_image", {}),
        "counts": counts,
        "prompt_length_profile": profile,
        "unmapped_to_production_fields": [
            "image_bytes",
            "sanitization_strategy",
            "validation_score",
            "validation_result",
            "openai_api_key",
        ],
    }


def _build_plate_prompt_compatibility_report(
    *, plan: dict, w6_plan: dict,
    production_diff_empty: bool, db_write_count: int,
    image_import_seen: bool, prev_run_id: str, missing_inputs: List[str],
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}
    plates = plan.get("plate_prompts", {}) or {}
    w6_payloads = w6_plan.get("payloads", {}) or {}

    inv["w6_inputs_present"] = {
        "pass": len(missing_inputs) == 0,
        "detail": {"missing": missing_inputs, "prev_run": prev_run_id},
    }

    expected = set(w6_payloads.keys())
    actual = set(plates.keys())
    inv["all_plate_prompts_covered"] = {
        "pass": bool(expected) and expected == actual,
        "detail": {
            "expected_count": len(expected),
            "actual_count": len(actual),
            "missing": sorted(expected - actual)[:5],
            "extra": sorted(actual - expected)[:5],
        },
    }

    backend = plan.get("image_generation_backend")
    wrong_model = [
        bg for bg, p in plates.items()
        if p.get("image_model") != W6_IMAGE_BACKEND
    ]
    inv["plate_model_is_gpt_image_2"] = {
        "pass": backend == W6_IMAGE_BACKEND and len(wrong_model) == 0,
        "detail": {"plan_backend": backend, "violating_count": len(wrong_model)},
    }

    # api_call_shape carried exact
    shape_mismatches: List[str] = []
    for bg in plates:
        if bg not in w6_payloads:
            continue
        if plates[bg].get("api_call_shape") != w6_payloads[bg].get("api_call_shape"):
            shape_mismatches.append(bg)
    inv["api_call_shape_carried_exact"] = {
        "pass": len(shape_mismatches) == 0,
        "detail": {"violating": shape_mismatches[:5]},
    }

    # reference_images carried exact
    ref_mismatches: List[str] = []
    for bg in plates:
        if bg not in w6_payloads:
            continue
        if plates[bg].get("reference_images") != w6_payloads[bg].get("reference_images"):
            ref_mismatches.append(bg)
    inv["reference_images_carried_exact"] = {
        "pass": len(ref_mismatches) == 0,
        "detail": {"violating": ref_mismatches[:5]},
    }

    inv["render_batches_carried_exact"] = {
        "pass": plan.get("render_batches") == w6_plan.get("render_batches"),
        "detail": {
            "w7_batches": len(plan.get("render_batches") or []),
            "w6_batches": len(w6_plan.get("render_batches") or []),
        },
    }

    inv["shot_to_background_image_carried_exact"] = {
        "pass": plan.get("shot_to_background_image") == w6_plan.get("shot_to_background_image"),
        "detail": {
            "w7_count": len(plan.get("shot_to_background_image") or {}),
            "w6_count": len(w6_plan.get("shot_to_background_image") or {}),
        },
    }

    empty_prompts = [bg for bg, p in plates.items()
                     if not (p.get("plate_prompt_text") or "").strip()]
    inv["plate_prompt_not_empty"] = {
        "pass": len(empty_prompts) == 0,
        "detail": {"violating": empty_prompts[:5]},
    }

    over_max = [
        f"{bg}:{len(p.get('plate_prompt_text') or '')}"
        for bg, p in plates.items()
        if len(p.get("plate_prompt_text") or "") > W7_PROMPT_HARD_MAX
    ]
    inv["plate_prompt_length_not_over_6000"] = {
        "pass": len(over_max) == 0,
        "detail": {"violating": over_max[:5], "hard_max": W7_PROMPT_HARD_MAX},
    }

    # W7b — high-risk group variants must use weak reference strength.
    # base plates in any group are 'none'; variants in high-risk groups are 'weak'.
    high_risk_violations: List[str] = []
    guides = plan.get("base_plate_group_guides") or {}
    for bg, p in plates.items():
        group_risk = (p.get("group_risk_level") or "").lower()
        if group_risk != "high":
            continue
        role = p.get("prompt_role")
        ref_strength = p.get("reference_strength")
        if role in ("strict_variant_delta", "weak_variant_delta") and ref_strength != "weak":
            high_risk_violations.append(f"{bg}:role={role}:ref_strength={ref_strength}")
        if role == "base_plate" and ref_strength != "none":
            high_risk_violations.append(f"{bg}:base_role_must_have_none_ref:{ref_strength}")
    # group-level strategy: every high-risk group's guide must declare weak_reference_variants
    for gid, g in guides.items():
        if (g.get("risk_level_from_w6") or "").lower() == "high":
            if g.get("base_prompt_strategy") != "weak_reference_variants":
                high_risk_violations.append(
                    f"{gid}:group_strategy={g.get('base_prompt_strategy')}"
                )
    inv["high_risk_groups_use_weak_reference_strategy"] = {
        "pass": len(high_risk_violations) == 0,
        "detail": {"violating": high_risk_violations[:5]},
    }

    banned_keys = {"needs_user_decision", "manual_review_required", "rollup",
                   "pending_human_decision", "human_review", "needs_approval"}
    bad_decision: List[str] = []
    for bg, p in plates.items():
        if any(bk in p for bk in banned_keys):
            bad_decision.append(bg)
    for gid, g in (plan.get("base_plate_group_guides") or {}).items():
        if any(bk in g for bk in banned_keys):
            bad_decision.append(gid)
    inv["no_human_decision_field"] = {
        "pass": len(bad_decision) == 0,
        "detail": {"violating": bad_decision[:5]},
    }

    inv["no_image_api_call"] = {
        "pass": not image_import_seen,
        "detail": "no openai.images.generate/edit / gemini_image_client / fal import",
    }
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }
    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


def _render_w7_html(run_meta: dict, plan: dict, report: dict, run_dir: Path) -> None:
    def esc(x):
        return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    counts = plan.get("counts", {}) or {}
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td class=\"{'pass' if v['pass'] else 'fail'}\">{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    sample_plate_items = list((plan.get("plate_prompts") or {}).items())[:2]
    sample_plates = {k: v for k, v in sample_plate_items}
    sample_guide_items = list((plan.get("base_plate_group_guides") or {}).items())[:2]
    sample_guides = {k: v for k, v in sample_guide_items}
    prof = plan.get("prompt_length_profile", {}) or {}
    val_errors = plan.get("validation_errors") or []
    val_block = (
        "<h2>validation errors</h2><pre>" + esc("\n".join(val_errors)) + "</pre>"
    ) if val_errors else ""
    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice W7 {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em;}}
table{{border-collapse:collapse;margin-bottom:1em}} td,th{{border:1px solid #ccc;padding:4px 8px}}
.metric{{display:inline-block;margin:0 1em 1em 0;padding:1em;border:1px solid #ddd;border-radius:6px}}
.fail{{color:#b00}} .pass{{color:#080}}</style></head>
<body>
<h1>background_pipeline_slice W7 — {esc(run_meta.get('run_id'))}</h1>
<p>stage: <b>{esc(run_meta.get('stage'))}</b> | plan_version: <b>{esc(run_meta.get('plan_version'))}</b>
| run_status: <b class=\"{'fail' if run_meta.get('run_status')!='succeeded' else 'pass'}\">{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}
| derived_from: {esc(run_meta.get('derived_from'))}
| image_backend: <b>{esc(run_meta.get('image_generation_backend'))}</b>
| model_used: <b>{esc(run_meta.get('model_used'))}</b>
| generation_status: <b>{esc(plan.get('generation_status'))}</b></p>
<div>
<div class=\"metric\"><div>plate prompts</div><b>{counts.get('plate_prompts', 0)}</b></div>
<div class=\"metric\"><div>groups</div><b>{counts.get('groups', 0)}</b></div>
<div class=\"metric\"><div>base_plate</div><b>{counts.get('role_base_plate', 0)}</b></div>
<div class=\"metric\"><div>strict_variant</div><b>{counts.get('role_strict_variant_delta', 0)}</b></div>
<div class=\"metric\"><div>weak_variant</div><b>{counts.get('role_weak_variant_delta', 0)}</b></div>
<div class=\"metric\"><div>fallback</div><b>{counts.get('role_fallback_independent_plate', 0)}</b></div>
<div class=\"metric\"><div>ref none/strict/weak</div><b>{counts.get('reference_strength_none', 0)}/{counts.get('reference_strength_strict', 0)}/{counts.get('reference_strength_weak', 0)}</b></div>
<div class=\"metric\"><div>groups weak-strategy</div><b>{counts.get('groups_strategy_weak_reference_variants', 0)}</b></div>
<div class=\"metric\"><div>generated</div><b>{counts.get('status_generated', 0)}</b></div>
<div class=\"metric\"><div>placeholder</div><b>{counts.get('status_placeholder_dry_run', 0)}</b></div>
<div class=\"metric\"><div>validation_failed</div><b>{counts.get('status_validation_failed', 0)}</b></div>
<div class=\"metric\"><div>validation</div><b>{'PASS' if report and report.get('all_pass') else 'FAIL'}</b></div>
</div>
{val_block}
<h2>W7 Invariants</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Prompt length profile (advisory >2000, hard >6000)</h2>
<pre>{esc(json.dumps(prof, ensure_ascii=False, indent=2))}</pre>
<h2>Sample plate prompts (first 2)</h2><pre>{esc(json.dumps(sample_plates, ensure_ascii=False, indent=2))}</pre>
<h2>Sample base_plate_group_guides (first 2)</h2><pre>{esc(json.dumps(sample_guides, ensure_ascii=False, indent=2))}</pre>
<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
<details><summary>raw plate_prompt_plan (HTML preview only, slice 80000 chars; full JSON preserved)</summary><pre>{esc(json.dumps(plan, ensure_ascii=False, indent=2))[:80000]}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


def _w7_main(args, run_dir: Path, run_id: str) -> int:
    """W7 stage — derive plate prompt plan from a prior W6 success run."""
    global _DB_WRITE_COUNT
    prev_run_dir = Path(args.derive_plate_prompts_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir
    artifacts = _load_w6_artifacts(prev_run_dir)
    missing = artifacts.get("_missing", [])
    outputs: List[str] = []
    failed: List[str] = []
    run_status = "succeeded"
    exit_code = 0
    model_used = None

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W7_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model": args.model if args.generate else None,
        "model_used": model_used,
        "image_generation_backend": W6_IMAGE_BACKEND,
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed,
    }

    if missing:
        failed.append("w6_inputs_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    w6_plan = artifacts["plan"]

    if args.generate:
        # backend/.env populates GEMINI_API_KEY when invoked outside the W1-W3 flow.
        _load_backend_env()
        # W7c — resolve upstream source_bundle via derived_from chain (W6 -> W5 -> W4 -> W3),
        # then build per-group source context. Failure is fail-closed.
        source_context: Optional[Dict[str, dict]] = None
        source_chain: List[str] = []
        try:
            resolved = _resolve_source_bundle_via_derived_from_chain(prev_run_dir)
            source_bundle = resolved["source_bundle"]
            source_chain = resolved["chain"]
            source_context = _build_per_group_source_context(w6_plan, source_bundle)
        except FileNotFoundError as exc:
            plan = _build_placeholder_plate_plan(w6_plan)
            plan["generation_status"] = "validation_failed"
            plan["validation_errors"] = [f"source_bundle_resolve_failed: {exc!s}"[:400]]
            failed.append("source_bundle_resolve_failed")
            source_context = None

        if source_context is not None:
            try:
                plan = _generate_llm_plate_plan(
                    w6_plan, model=args.model, source_context=source_context,
                )
                model_used = args.model
                plan["source_bundle_chain"] = source_chain
            except Exception as exc:
                plan = _build_placeholder_plate_plan(w6_plan)
                plan["generation_status"] = "validation_failed"
                plan["validation_errors"] = [f"llm_call_failed: {exc!s}"[:400]]
                plan["source_bundle_chain"] = source_chain
                failed.append("llm_call_failed")
    else:
        plan = _build_placeholder_plate_plan(w6_plan)

    (run_dir / "background_plate_prompt_plan.json").write_text(
        json.dumps(plan, ensure_ascii=False, indent=2)
    )
    outputs.append("background_plate_prompt_plan.json")

    report = _build_plate_prompt_compatibility_report(
        plan=plan, w6_plan=w6_plan,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_import_seen=_check_image_imports_present(),
        prev_run_id=prev_run_dir.name,
        missing_inputs=missing,
    )
    (run_dir / "plate_prompt_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("plate_prompt_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed:
            failed.append(name)
    if plan.get("generation_status") == "validation_failed" and "llm_validation_failed" not in failed:
        failed.append("llm_validation_failed")
    if failed:
        run_status = "validation_failed"
        exit_code = 1

    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed
    run_meta["model_used"] = plan.get("model_used")
    run_meta["outputs"] = outputs
    _render_w7_html(run_meta, plan, report, run_dir)
    outputs.append("index.html")
    run_meta["outputs"] = outputs
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    _maybe_print_imports(args)
    return exit_code


# ─────────────────────────────────────────────────────────────────────────────
# W8 — actual gpt-image-2 image generation (narrow, opt-in via --generate-images)
# ─────────────────────────────────────────────────────────────────────────────

W8_STAGE = "w8_image_generation"
W8_IMAGE_MODEL = "gpt-image-2"
W8_DEFAULT_SIZE = "1024x1024"
W8_DEFAULT_QUALITY = "high"


def _load_w7_artifacts(prev_run_dir: Path) -> dict:
    """W8 input — load W7 success run artifacts read-only."""
    required = {
        "plan": "background_plate_prompt_plan.json",
        "report": "plate_prompt_compatibility_report.json",
        "run_meta": "run_meta.json",
    }
    out: Dict[str, Any] = {}
    missing: List[str] = []
    for key, fname in required.items():
        p = prev_run_dir / fname
        if not p.exists():
            missing.append(fname)
            continue
        out[key] = json.loads(p.read_text())
    out["_missing"] = missing
    out["_prev_run_id"] = prev_run_dir.name
    return out


def _parse_selected_bg_ids(arg_value: Optional[str]) -> Optional[List[str]]:
    if not arg_value:
        return None
    items = [s.strip() for s in arg_value.split(",")]
    return [s for s in items if s]


def _select_and_order_bg_ids(w7_plan: dict, selected: Optional[List[str]]) -> List[str]:
    """Order selected bg_ids by the W7 render_batches; selected==None means all."""
    plate_keys = list((w7_plan.get("plate_prompts") or {}).keys())
    batches = w7_plan.get("render_batches") or []
    if selected is None:
        wanted = set(plate_keys)
    else:
        wanted = {bg for bg in selected if bg in plate_keys}
    ordered: List[str] = []
    for batch in batches:
        for bg in batch:
            if bg in wanted and bg not in ordered:
                ordered.append(bg)
    # any wanted that did not appear in any batch (shouldn't happen with valid W7) append at end
    for bg in plate_keys:
        if bg in wanted and bg not in ordered:
            ordered.append(bg)
    return ordered


def _resolve_w8_openai_client():
    """W8 OpenAI client. api_key from os.environ (populated by _load_backend_env)."""
    from openai import OpenAI  # lazy; dry-run never imports
    api_key = os.environ.get("OPENAI_API_KEY")
    if not api_key:
        raise RuntimeError("missing OPENAI_API_KEY env var")
    timeout = float(os.environ.get("LLM_TIMEOUT_IMAGE_GEN", "600"))
    return OpenAI(api_key=api_key, timeout=timeout)


W8E_DEFAULT_AVOIDANCE_SUFFIX = (
    "Default-avoidance instruction: keep the generated space modest, narrow, "
    "and source-grounded; do not upgrade it into a spacious luxury, high-end "
    "minimalist, or bright expansive modern interior. Do not add adjacent "
    "areas or objects unless they are already requested by the per-bg prompt."
)

# W8f — generic framing prefix prepended to every per-bg image prompt. Tells
# the image model the artifact is a film-production set/background plate
# reference, not a real-world depiction. Generic policy; identical wording
# for every bg regardless of scenario.
W8F_FILM_SET_FRAMING_PREFIX = (
    "This image is a film-production background plate reference for cinematic "
    "set design. It depicts an empty stage/background environment for "
    "storyboarding and production design, not a real-world event or person. "
    "Render it as concept art for a film set, with no live characters present."
)


def _assemble_w8c_prompt(plate_prompt_text: str, group_guide: Optional[dict]) -> dict:
    """W8c/W8d/W8e — structural carry policy for source-derived group context.

    W8e policy: do NOT append the raw scale_and_default_avoidance value (or any
    other group_guide field) into the final prompt. Group-level source text
    carries its own object/area mentions that can leak into per-bg prompts in
    multi-space groups. Instead, when a group_guide is present, append a
    single generic default-avoidance suffix that states the policy in
    category-level wording only. The per-bg plate_prompt_text remains the
    local source-of-truth.

    Returns {"prompt_text_used": str, "carried_fields": [str],
             "applied_default_avoidance_policy": bool}.
    """
    # W8f — generic film-set framing prefix always present (when there is any
    # per-bg prompt text to wrap). Tells the image model this is a background
    # plate reference for cinematic set design, mitigating contextual NSFW
    # mis-classification on otherwise innocuous environmental descriptions.
    parts: List[str] = []
    applied_film_set_prefix = False
    if (plate_prompt_text or "").strip():
        parts.append(W8F_FILM_SET_FRAMING_PREFIX)
        applied_film_set_prefix = True
    parts.append(plate_prompt_text or "")
    applied_policy = False
    if isinstance(group_guide, dict):
        # The presence of a group_guide signals there is upstream source-derived
        # context for this group; we apply the generic policy suffix only.
        # No verbatim value carry to avoid multi-space subspace leakage.
        parts.append(W8E_DEFAULT_AVOIDANCE_SUFFIX)
        applied_policy = True
    return {
        "prompt_text_used": "\n\n".join(p for p in parts if p),
        # carried_fields stays empty under W8e: no group field value is pushed
        # into the final prompt. applied_default_avoidance_policy records that
        # the generic policy suffix was appended.
        "carried_fields": [],
        "applied_default_avoidance_policy": applied_policy,
        "applied_film_set_framing_prefix": applied_film_set_prefix,
    }


def _generate_image_for_bg(
    bg_id: str, plate_entry: dict, *,
    openai_client: Any, images_dir: Path,
    dry_run: bool, weak_reference_mode: str = "edit_with_ref",
    group_guide: Optional[dict] = None,
    subgroup_info: Optional[dict] = None,
) -> dict:
    """Generate a single bg's PNG. Returns per-bg result dict (always).

    weak_reference_mode (W8b):
      - 'edit_with_ref' (default): weak payloads still use images.edit with the
        declared reference PNG (W8 behavior).
      - 'prompt_only': weak payloads skip the reference image and use
        images.generate with plate_prompt_text alone. declared_reference_bg_ids
        is preserved in the result; reference_pngs_used stays empty; the
        effective_reference_mode marker records the policy switch.

    group_guide (W8c): optional group_guide dict (from the W7 plan). When
    present, source-derived constraint fields are appended verbatim to the
    per-bg prompt as a structural carry. The runtime never interprets the
    constraint words; it only joins JSON field values.
    """
    api_shape = plate_entry.get("api_call_shape") or {}
    declared_model = api_shape.get("model")
    plate_prompt_text_original = plate_entry.get("plate_prompt_text") or ""
    assembled = _assemble_w8c_prompt(plate_prompt_text_original, group_guide)
    prompt_text = assembled["prompt_text_used"]
    carried_fields = assembled["carried_fields"]
    applied_default_avoidance_policy = assembled.get(
        "applied_default_avoidance_policy", False
    )
    applied_film_set_framing_prefix = assembled.get(
        "applied_film_set_framing_prefix", False
    )
    reference_images = plate_entry.get("reference_images") or []
    declared_client_method = api_shape.get("client_method") or "images.generate"
    reference_strength = (plate_entry.get("reference_strength") or "").lower()

    declared_ref_bg_ids = [r.get("bg_id") for r in reference_images if isinstance(r, dict)]

    # W9 — same-subgroup direct parent override.
    # If subgroup_info tells us this bg has a direct parent in the SAME
    # image_continuity_subgroup, AND that parent is one of the declared refs,
    # we keep images.edit (state transition within the same physical subspace).
    # Otherwise we still respect --weak-reference-mode.
    same_subgroup_parent_bg = None
    if isinstance(subgroup_info, dict):
        chain = subgroup_info.get("generation_chain") or []
        pos = subgroup_info.get("position_in_chain")
        if isinstance(pos, int) and pos > 0:
            candidate = chain[pos - 1]
            if candidate in declared_ref_bg_ids:
                same_subgroup_parent_bg = candidate

    # W8b — decide whether weak refs are dropped entirely for this call.
    drop_refs_for_weak = (
        reference_strength == "weak"
        and weak_reference_mode == "prompt_only"
        and same_subgroup_parent_bg is None
    )

    if drop_refs_for_weak:
        effective_reference_mode = "prompt_only_weak"
        effective_client_method = "images.generate"
    elif same_subgroup_parent_bg is not None:
        effective_reference_mode = "subgroup_state_transition"
        effective_client_method = "images.edit"
    else:
        effective_reference_mode = "edit_with_ref" if declared_ref_bg_ids else "none"
        effective_client_method = declared_client_method

    result: Dict[str, Any] = {
        "bg_id": bg_id,
        "status": "pending",
        "model_used": None,
        "model_requested": declared_model,
        "subgroup_id": (subgroup_info or {}).get("subgroup_id") if isinstance(subgroup_info, dict) else None,
        "same_subgroup_parent_bg": same_subgroup_parent_bg,
        "client_method_declared": declared_client_method,
        "client_method_effective": effective_client_method,
        # W8c: prompt_text_used = final prompt sent to gpt-image-2 (plate
        # prompt + carried group-guide fields). plate_prompt_text_original
        # preserves the original W7 plate prompt for audit.
        "prompt_text_used": prompt_text,
        "plate_prompt_text_original": plate_prompt_text_original,
        "carried_group_guide_fields": carried_fields,
        "applied_default_avoidance_policy": applied_default_avoidance_policy,
        "applied_film_set_framing_prefix": applied_film_set_framing_prefix,
        "reference_strength": reference_strength,
        "weak_reference_mode": weak_reference_mode,
        "effective_reference_mode": effective_reference_mode,
        "declared_reference_bg_ids": declared_ref_bg_ids,
        # legacy alias for HTML/back-compat; mirrors declared list for backwards reads.
        "reference_bg_ids": declared_ref_bg_ids,
        "reference_pngs_used": [],
        "png_path": None,
        "size": W8_DEFAULT_SIZE,
        "quality": W8_DEFAULT_QUALITY,
        "n": 1,
        "error": None,
    }

    # Enforce gpt-image-2 from the upstream plate plan; refuse otherwise.
    if declared_model != W8_IMAGE_MODEL:
        result["status"] = "validation_failed"
        result["error"] = f"plate_plan_declared_model_not_gpt_image_2: {declared_model}"
        return result

    if dry_run:
        # Dry-run: do not check reference PNG existence (nothing has been
        # generated yet; missing PNGs are expected). Just echo the declared refs.
        result["status"] = "dry_run_skipped"
        return result

    # Generate path: resolve reference PNGs only when effective policy uses them.
    ref_paths: List[Path] = []
    if effective_reference_mode == "subgroup_state_transition":
        # Use only the same-subgroup direct parent PNG; ignore other declared
        # refs (e.g., loc-group base in a different subspace).
        ref_path = images_dir / f"{same_subgroup_parent_bg}.png"
        if not ref_path.exists():
            result["status"] = "blocked_missing_reference"
            result["error"] = f"reference_png_missing: {same_subgroup_parent_bg}.png"
            return result
        ref_paths.append(ref_path)
    elif not drop_refs_for_weak:
        for ref in reference_images:
            if not isinstance(ref, dict):
                continue
            ref_bg = ref.get("bg_id")
            if not ref_bg:
                continue
            ref_path = images_dir / f"{ref_bg}.png"
            if not ref_path.exists():
                result["status"] = "blocked_missing_reference"
                result["error"] = f"reference_png_missing: {ref_bg}.png"
                return result
            ref_paths.append(ref_path)
    result["reference_pngs_used"] = [str(p) for p in ref_paths]

    # Actual gpt-image-2 call. Sequential, n=1.
    import base64
    out_path = images_dir / f"{bg_id}.png"
    try:
        if ref_paths:
            if len(ref_paths) >= 2:
                import contextlib as _cl
                with _cl.ExitStack() as stack:
                    files = [stack.enter_context(p.open("rb")) for p in ref_paths]
                    resp = openai_client.images.edit(
                        model=W8_IMAGE_MODEL,
                        image=files,
                        prompt=prompt_text,
                        size=W8_DEFAULT_SIZE,
                        quality=W8_DEFAULT_QUALITY,
                        n=1,
                    )
            else:
                with ref_paths[0].open("rb") as f:
                    resp = openai_client.images.edit(
                        model=W8_IMAGE_MODEL,
                        image=f,
                        prompt=prompt_text,
                        size=W8_DEFAULT_SIZE,
                        quality=W8_DEFAULT_QUALITY,
                        n=1,
                    )
        else:
            resp = openai_client.images.generate(
                model=W8_IMAGE_MODEL,
                prompt=prompt_text,
                size=W8_DEFAULT_SIZE,
                quality=W8_DEFAULT_QUALITY,
                n=1,
            )
        b64 = resp.data[0].b64_json if resp and resp.data else None
        if not b64:
            raise RuntimeError("empty b64_json from gpt-image-2 response")
        out_path.write_bytes(base64.b64decode(b64))
        result["status"] = "generated"
        result["model_used"] = W8_IMAGE_MODEL
        result["png_path"] = str(out_path)
    except Exception as exc:
        result["status"] = "generation_failed"
        result["error"] = f"{type(exc).__name__}: {str(exc)[:400]}"
    return result


def _build_image_generation_compatibility_report(
    *, results: List[dict], plan: dict, selected_bg_ids: List[str],
    production_diff_empty: bool, db_write_count: int,
    image_call_made: bool, dry_run_requested: bool,
    missing_inputs: List[str], prev_run_id: str,
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}
    plate_keys = set((plan.get("plate_prompts") or {}).keys())
    by_bg = {r["bg_id"]: r for r in results}

    inv["w7_inputs_present"] = {
        "pass": len(missing_inputs) == 0,
        "detail": {"missing": missing_inputs, "prev_run": prev_run_id},
    }

    inv["selected_bg_ids_subset_of_plan"] = {
        "pass": all(bg in plate_keys for bg in selected_bg_ids),
        "detail": {
            "extra": sorted(set(selected_bg_ids) - plate_keys)[:5],
            "selected_count": len(selected_bg_ids),
        },
    }

    # Every attempted bg's plate plan declared gpt-image-2; result echoes it.
    model_violations = [
        r["bg_id"] for r in results if r.get("model_requested") != W8_IMAGE_MODEL
    ]
    inv["model_is_gpt_image_2_for_all_attempts"] = {
        "pass": len(model_violations) == 0,
        "detail": {"violating": model_violations[:5]},
    }

    # generation_order_respects_dependency (W8b — based on EFFECTIVE refs).
    # When a payload's effective policy still consumes a reference PNG, the
    # referenced bg must appear earlier in selected_bg_ids. Payloads whose
    # effective_reference_mode is 'prompt_only_weak' have NO effective deps.
    order_index = {bg: i for i, bg in enumerate(selected_bg_ids)}
    dep_violations: List[str] = []
    for r in results:
        bg = r["bg_id"]
        effective_mode = r.get("effective_reference_mode")
        if effective_mode in ("prompt_only_weak", "none"):
            continue  # no effective refs
        if effective_mode == "subgroup_state_transition":
            # only the same-subgroup parent counts as effective dep
            same_parent = r.get("same_subgroup_parent_bg")
            if same_parent and same_parent in order_index and order_index[same_parent] < order_index.get(bg, 0):
                continue
            if same_parent:
                dep_violations.append(
                    f"{bg}_subgroup_parent_{same_parent}_position_invalid"
                )
            continue
        for ref_bg in (r.get("declared_reference_bg_ids") or []):
            if not ref_bg:
                continue
            if ref_bg not in order_index:
                dep_violations.append(f"{bg}:ref_{ref_bg}_not_in_selection")
                continue
            if order_index[ref_bg] >= order_index[bg]:
                dep_violations.append(
                    f"{bg}(idx={order_index[bg]})_refs_{ref_bg}(idx={order_index[ref_bg]})"
                )
    inv["generation_order_respects_dependency"] = {
        "pass": len(dep_violations) == 0,
        "detail": {"violating": dep_violations[:5]},
    }

    # references_resolved_when_needed: if --generate-images requested, no
    # payload that actually consumes a reference should be in
    # blocked_missing_reference. prompt_only_weak skips ref entirely so it
    # cannot be blocked by missing ref PNG.
    blocked = [r["bg_id"] for r in results if r["status"] == "blocked_missing_reference"]
    if dry_run_requested:
        inv["references_resolved_when_needed"] = {
            "pass": True,
            "detail": "dry-run: not enforced",
        }
    else:
        inv["references_resolved_when_needed"] = {
            "pass": len(blocked) == 0,
            "detail": {"blocked": blocked[:5]},
        }

    # all_emitted_pngs_exist for generated entries
    generated_missing = [
        r["bg_id"] for r in results
        if r["status"] == "generated" and (not r.get("png_path") or not Path(r["png_path"]).exists())
    ]
    inv["all_emitted_pngs_exist"] = {
        "pass": len(generated_missing) == 0,
        "detail": {"missing": generated_missing[:5]},
    }

    # generation_failed and validation_failed propagate as failures (advisory in dry-run)
    failed = [r["bg_id"] for r in results
              if r["status"] in ("generation_failed", "validation_failed")]
    inv["no_unexpected_generation_failures"] = {
        "pass": len(failed) == 0,
        "detail": {"failed": failed[:5]},
    }

    # ban human-decision fields in results
    banned_keys = {"needs_user_decision", "manual_review_required", "rollup",
                   "pending_human_decision", "human_review", "needs_approval"}
    bad_decision = [r["bg_id"] for r in results if any(bk in r for bk in banned_keys)]
    inv["no_human_decision_field"] = {
        "pass": len(bad_decision) == 0,
        "detail": {"violating": bad_decision[:5]},
    }

    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


def _render_w8_html(run_meta: dict, *, plan: dict, results: List[dict],
                    report: dict, run_dir: Path) -> None:
    def esc(x):
        return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td class=\"{'pass' if v['pass'] else 'fail'}\">{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    plate_lookup = plan.get("plate_prompts") or {}
    cards: List[str] = []
    for r in results:
        bg_id = r["bg_id"]
        plate = plate_lookup.get(bg_id) or {}
        status = r.get("status", "")
        png_rel = f"images/{bg_id}.png" if status == "generated" else None
        declared_refs = r.get("declared_reference_bg_ids") or r.get("reference_bg_ids") or []
        effective_refs = [Path(p).stem for p in (r.get("reference_pngs_used") or [])]
        effective_mode = r.get("effective_reference_mode", "")
        client_method_effective = r.get("client_method_effective", r.get("client_method", ""))
        role = plate.get("prompt_role", "")
        ref_strength = plate.get("reference_strength", "")
        prompt_used = r.get("prompt_text_used", "")
        error = r.get("error")
        ref_blocks: List[str] = []
        if declared_refs:
            ref_blocks.append(
                "<div class='refs'>declared refs: " + " ".join(
                    f"<a href='#bg-{esc(rb)}'>{esc(rb)}</a>" for rb in declared_refs
                ) + "</div>"
            )
        ref_blocks.append(
            "<div class='refs'>effective refs: " + (
                " ".join(f"<a href='#bg-{esc(rb)}'>{esc(rb)}</a>" for rb in effective_refs)
                if effective_refs else "(none)"
            )
            + f" | mode={esc(effective_mode)} | call={esc(client_method_effective)}</div>"
        )
        ref_thumbs = "".join(ref_blocks)
        # status-driven main visual
        if png_rel:
            visual = f"<img class='thumb' src='{esc(png_rel)}' alt='{esc(bg_id)}'>"
        elif status == "blocked_missing_reference":
            visual = "<div class='blocked'>blocked_missing_reference</div>"
        elif status == "dry_run_skipped":
            visual = "<div class='dry'>dry_run_skipped</div>"
        elif status == "generation_failed":
            visual = f"<div class='fail'>generation_failed</div>"
        elif status == "validation_failed":
            visual = f"<div class='fail'>validation_failed</div>"
        else:
            visual = f"<div class='unknown'>{esc(status)}</div>"

        cards.append(f"""<div class='card' id='bg-{esc(bg_id)}'>
  <div class='cardhead'>
    <b>{esc(bg_id)}</b>
    <span class='role'>{esc(role)}</span>
    <span class='ref'>ref_strength={esc(ref_strength)}</span>
    <span class='status status-{esc(status)}'>{esc(status)}</span>
  </div>
  {visual}
  {ref_thumbs}
  <details><summary>plate_prompt_text used</summary><pre>{esc(prompt_used)}</pre></details>
  {('<div class=err>error: ' + esc(str(error)) + '</div>') if error else ''}
</div>""")
    cards_html = "\n".join(cards)
    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice W8 {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em;background:#111;color:#eee;}}
table{{border-collapse:collapse}} td,th{{border:1px solid #444;padding:4px 8px}}
.metric{{display:inline-block;margin:0 1em 1em 0;padding:1em;border:1px solid #444;border-radius:6px}}
.fail{{color:#f66}} .pass{{color:#7c7}}
.card{{border:1px solid #444;border-radius:6px;padding:1em;margin-bottom:1.5em;background:#1a1a1a;max-width:1100px}}
.cardhead{{margin-bottom:0.6em;font-size:1.05em}}
.cardhead .role,.cardhead .ref,.cardhead .status{{margin-left:0.6em;padding:0.1em 0.5em;border:1px solid #555;border-radius:4px;font-size:0.85em}}
.thumb{{max-width:1024px;width:100%;border:1px solid #666;border-radius:4px;display:block;margin:0.5em 0}}
.refs{{margin:0.5em 0;font-size:0.9em;color:#aaa}}
.refs a{{color:#9cf}}
.blocked,.dry,.unknown{{padding:1em;background:#222;border:1px dashed #777;border-radius:4px;color:#ccc;text-align:center;font-style:italic;margin:0.5em 0}}
.err{{color:#f66;margin-top:0.4em}}
pre{{white-space:pre-wrap;color:#ccc;font-size:0.85em;background:#0a0a0a;padding:0.6em;border-radius:4px}}
details summary{{cursor:pointer;color:#9cf}}
.status-generated{{color:#7c7}}.status-blocked_missing_reference{{color:#fa3}}.status-dry_run_skipped{{color:#9cf}}.status-generation_failed,.status-validation_failed{{color:#f66}}
</style></head>
<body>
<h1>background_pipeline_slice W8 — {esc(run_meta.get('run_id'))}</h1>
<p>stage: <b>{esc(run_meta.get('stage'))}</b> | run_status: <b class=\"{'fail' if run_meta.get('run_status')!='succeeded' else 'pass'}\">{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}
| derived_from: {esc(run_meta.get('derived_from'))}
| image_model: <b>{esc(W8_IMAGE_MODEL)}</b>
| generate_images: <b>{esc(run_meta.get('args', {}).get('generate_images'))}</b></p>
<h2>Selected bg_ids (in generation order)</h2>
<p>{', '.join(esc(r['bg_id']) + ' [' + esc(r['status']) + ']' for r in results)}</p>
<h2>W8 invariants</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Images</h2>
{cards_html}
<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


def _w8_main(args, run_dir: Path, run_id: str) -> int:
    """W8 stage — actual gpt-image-2 generation for a subset of W7 plate prompts."""
    global _DB_WRITE_COUNT
    prev_run_dir = Path(args.derive_image_generation_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir
    artifacts = _load_w7_artifacts(prev_run_dir)
    missing = artifacts.get("_missing", [])
    outputs: List[str] = []
    failed: List[str] = []
    run_status = "succeeded"
    exit_code = 0
    dry_run = not bool(args.generate_images)

    images_dir = run_dir / "images"
    images_dir.mkdir(parents=True, exist_ok=True)

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W8_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "image_model": W8_IMAGE_MODEL,
        "image_generation_backend": W8_IMAGE_MODEL,
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed,
    }

    if missing:
        failed.append("w7_inputs_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    w7_plan = artifacts["plan"]
    selected = _parse_selected_bg_ids(args.selected_bg_ids)
    selected_bg_ids = _select_and_order_bg_ids(w7_plan, selected)

    openai_client = None
    if not dry_run:
        _load_backend_env()
        try:
            openai_client = _resolve_w8_openai_client()
        except Exception as exc:
            failed.append("openai_client_unavailable")
            run_status = "validation_failed"
            exit_code = 1
            run_meta["run_status"] = run_status
            run_meta["exit_code"] = exit_code
            run_meta["failed_invariants"] = failed
            (run_dir / "run_meta.json").write_text(
                json.dumps({**run_meta, "error": str(exc)[:400]}, ensure_ascii=False, indent=2)
            )
            return exit_code

    results: List[dict] = []
    image_call_made = False
    plates = w7_plan.get("plate_prompts") or {}
    group_guides = w7_plan.get("base_plate_group_guides") or {}
    # bg_id -> group_id from W7 plan groups (exact id matching)
    bg_to_group_id: Dict[str, str] = {}
    for g in (w7_plan.get("groups") or []):
        gid = g.get("group_id")
        for m in (g.get("member_bg_ids") or []):
            bg_to_group_id[m] = gid

    # W9 — optional W8h subgroup info
    bg_to_subgroup_info: Dict[str, dict] = {}
    continuity_from_dir = None
    if getattr(args, "continuity_from", None):
        cont_path = Path(args.continuity_from)
        if not cont_path.is_absolute():
            cont_path = Path.cwd() / cont_path
        try:
            continuity = _load_continuity_subgroups(cont_path)
            bg_to_subgroup_info = _build_bg_to_subgroup_index(continuity)
            continuity_from_dir = cont_path.name
        except FileNotFoundError as exc:
            failed.append("continuity_subgroups_missing")
            run_meta["error"] = str(exc)[:300]
            run_status = "validation_failed"
            exit_code = 1

    for bg in selected_bg_ids:
        entry = plates.get(bg) or {}
        gid = bg_to_group_id.get(bg)
        guide = group_guides.get(gid) if gid else None
        sub_info = bg_to_subgroup_info.get(bg)
        r = _generate_image_for_bg(
            bg, entry, openai_client=openai_client,
            images_dir=images_dir, dry_run=dry_run,
            weak_reference_mode=args.weak_reference_mode,
            group_guide=guide,
            subgroup_info=sub_info,
        )
        results.append(r)
        if r["status"] == "generated":
            image_call_made = True
    if continuity_from_dir:
        run_meta["continuity_from"] = continuity_from_dir

    out_data = {
        "schema_version": 1,
        "stage": W8_STAGE,
        "plan_version": PLAN_VERSION,
        "image_model": W8_IMAGE_MODEL,
        "dry_run": dry_run,
        "derived_from": prev_run_dir.name,
        "selected_bg_ids": selected_bg_ids,
        "results": results,
    }
    (run_dir / "background_image_generation_result.json").write_text(
        json.dumps(out_data, ensure_ascii=False, indent=2)
    )
    outputs.append("background_image_generation_result.json")

    report = _build_image_generation_compatibility_report(
        results=results, plan=w7_plan, selected_bg_ids=selected_bg_ids,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_call_made=image_call_made, dry_run_requested=dry_run,
        missing_inputs=missing, prev_run_id=prev_run_dir.name,
    )
    (run_dir / "image_generation_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("image_generation_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed:
            failed.append(name)
    if failed:
        run_status = "validation_failed"
        exit_code = 1

    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed
    run_meta["outputs"] = outputs
    _render_w8_html(run_meta, plan=w7_plan, results=results, report=report, run_dir=run_dir)
    outputs.append("index.html")
    run_meta["outputs"] = outputs
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    _maybe_print_imports(args)
    return exit_code


# ─────────────────────────────────────────────────────────────────────────────
# W8h — image_continuity_subgroups LLM stage
# ─────────────────────────────────────────────────────────────────────────────

W8H_STAGE = "w8h_image_continuity_subgroups"

LLM_IMAGE_CONTINUITY_SCHEMA = {
    "type": "object",
    "required": ["image_continuity_subgroups"],
    "properties": {
        "image_continuity_subgroups": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["subgroup_id", "member_bg_ids", "base_bg_id",
                             "generation_chain", "reason_brief", "confidence"],
                "properties": {
                    "subgroup_id": {"type": "string"},
                    "member_bg_ids": {"type": "array", "items": {"type": "string"}},
                    "base_bg_id": {"type": "string"},
                    "generation_chain": {"type": "array", "items": {"type": "string"}},
                    "reason_brief": {"type": "string"},
                    "confidence": {"type": "string"},
                },
            },
        },
    },
}


def _load_w3_adapter_from_chain(prev_run_dir: Path) -> dict:
    """Walk derived_from chain from W7 → ... → W3, return production_adapter_plan.json content."""
    parent_dir = prev_run_dir.parent
    cur = prev_run_dir
    for _ in range(W7C_CHAIN_MAX_HOPS + 1):
        adapter_path = cur / "production_adapter_plan.json"
        if adapter_path.exists():
            return json.loads(adapter_path.read_text())
        rm_path = cur / "run_meta.json"
        if not rm_path.exists():
            raise FileNotFoundError(f"adapter chain broken at {cur.name}")
        rm = json.loads(rm_path.read_text())
        prev_id = rm.get("derived_from")
        if not prev_id:
            raise FileNotFoundError(
                f"adapter chain terminated at {cur.name} without production_adapter_plan.json"
            )
        cur = parent_dir / prev_id
        if not cur.exists():
            raise FileNotFoundError(f"adapter chain dir missing: {prev_id}")
    raise FileNotFoundError("adapter chain exceeded max hops")


def _build_continuity_llm_input(w7_plan: dict, source_bundle: dict,
                                adapter_plan: dict) -> dict:
    """Compact view fed to the continuity LLM. Generic carry of existing JSON
    fields; no semantic interpretation."""
    plates = w7_plan.get("plate_prompts") or {}
    groups = w7_plan.get("groups") or []
    catalog = adapter_plan.get("background_catalog") or {}
    bg_to_group = {}
    for g in groups:
        gid = g.get("group_id")
        for m in (g.get("member_bg_ids") or []):
            bg_to_group[m] = gid

    bg_views: List[dict] = []
    for bg_id, p in plates.items():
        catalog_entry = catalog.get(bg_id) or {}
        bg_views.append({
            "bg_id": bg_id,
            "group_id": bg_to_group.get(bg_id),
            "plate_prompt_text": p.get("plate_prompt_text"),
            "prompt_role": p.get("prompt_role"),
            "reference_strength": p.get("reference_strength"),
            "render_mode": p.get("render_mode"),
            "applies_to_shots": p.get("applies_to_shots"),
            "depends_on_bg": catalog_entry.get("depends_on_bg"),
            "sub_location_label": catalog_entry.get("sub_location_label"),
            "state_label_raw": catalog_entry.get("state_label_raw"),
            "loc_id": catalog_entry.get("loc_id"),
            "space_key": catalog_entry.get("space_key"),
            "time_phase": catalog_entry.get("time_phase"),
            "state_class": catalog_entry.get("state_class"),
        })
    return {
        "background_entries": bg_views,
        "groups": [
            {"group_id": g.get("group_id"), "base_bg_id": g.get("base_bg_id"),
             "member_bg_ids": g.get("member_bg_ids", []),
             "loc_id": g.get("loc_id"), "space_key": g.get("space_key")}
            for g in groups
        ],
        "json_schema": LLM_IMAGE_CONTINUITY_SCHEMA,
    }


def _generate_image_continuity_via_llm(llm_input: dict, *, model: str) -> dict:
    """--generate path. Asks the LLM to partition bg_ids into physical-subspace
    subgroups so that same-room state transitions keep a direct parent
    reference. Deterministic checker (caller) validates set partition shape."""
    import litellm

    system_prompt = (
        "You partition background image entries into image_continuity_subgroups for "
        "downstream image-generation. The partition must be FINER than the loc_id / "
        "group_id partition; same loc_id frequently spans multiple distinct physical "
        "subspaces (rooms, halves of a multi-space dwelling, interior vs exterior, "
        "different sides of a structure) and these must end up in DIFFERENT "
        "subgroups. "
        "Two bg_ids belong to the SAME subgroup ONLY when they depict the same "
        "physical subspace/room/scene (architecture-level continuity), differing "
        "only by state, time, lighting, or material/surface transformation — i.e. a "
        "downstream model could plausibly produce one by editing the other while "
        "keeping the same walls, doorways, windows, and floor plan. "
        "Two bg_ids belong to DIFFERENT subgroups when their sub_location_label, "
        "state_label_raw, applies_to_shots, or plate_prompt_text imply distinct "
        "rooms, distinct interior/exterior zones, or distinct architectural areas — "
        "even if they share the same loc_id, group_id, or depends_on_bg parent. A "
        "depends_on_bg edge in the input adapter is NOT proof of same subspace; it "
        "only constrains downstream rendering order. Treat depends_on_bg as a soft "
        "hint only. "
        "If you are unsure whether two bg_ids share a physical subspace, default to "
        "SPLITTING them into separate subgroups (single-member subgroup with "
        "confidence='medium') rather than merging. A multi-space dwelling typically "
        "yields several small subgroups, not one big one. "
        "Use sub_location_label, state_label_raw, plate_prompt_text wording, "
        "applies_to_shots context, and depends_on_bg only as evidence; do not invent "
        "subgroups for hypothetical spaces not present in the input. Every input "
        "bg_id must appear in EXACTLY one subgroup. Within each subgroup, set "
        "base_bg_id to the bg most likely to be generated first (typically the "
        "independent/master_base in that subspace; if none exists, pick the most "
        "state-neutral member) and emit generation_chain as a topologically sorted "
        "subset that starts with base_bg_id. Generation_chain must be acyclic; do "
        "not list a member twice. reason_brief is at most two short sentences and "
        "must justify why members are or are NOT split: cite sub_location_label or "
        "the deciding evidence. confidence is 'high', 'medium', or 'low'. Output "
        "strict JSON matching the provided json_schema; no extra keys, no commentary."
    )
    user_prompt = json.dumps(
        {"background_entries": llm_input["background_entries"],
         "groups": llm_input["groups"],
         "json_schema": llm_input["json_schema"]},
        ensure_ascii=False,
    )
    routed = (
        model
        if model.startswith("gemini/") or not model.lower().startswith("gemini")
        else f"gemini/{model}"
    )
    if not (os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")):
        raise RuntimeError("missing GEMINI_API_KEY / GOOGLE_API_KEY env var")
    resp = litellm.completion(
        model=routed,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        response_format={"type": "json_object"},
    )
    raw_text = resp.choices[0].message.content
    decoder = json.JSONDecoder()
    stripped = raw_text.lstrip()
    parsed, _ = decoder.raw_decode(stripped)
    import jsonschema
    jsonschema.validate(parsed, LLM_IMAGE_CONTINUITY_SCHEMA)
    return parsed


def _build_placeholder_continuity(w7_plan: dict, adapter_plan: dict) -> dict:
    """Dry-run — one subgroup per bg (no chain). status=placeholder_dry_run."""
    plates = w7_plan.get("plate_prompts") or {}
    subgroups: List[dict] = []
    for bg_id in plates:
        subgroups.append({
            "subgroup_id": f"placeholder_{bg_id}",
            "member_bg_ids": [bg_id],
            "base_bg_id": bg_id,
            "generation_chain": [bg_id],
            "reason_brief": "placeholder_dry_run",
            "confidence": "low",
        })
    return {"image_continuity_subgroups": subgroups}


def _build_continuity_compatibility_report(
    *, continuity: dict, w7_plan: dict,
    production_diff_empty: bool, db_write_count: int,
    image_import_seen: bool, prev_run_id: str, missing_inputs: List[str],
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}
    subgroups = continuity.get("image_continuity_subgroups") or []
    plate_keys = set((w7_plan.get("plate_prompts") or {}).keys())

    inv["w7_inputs_present"] = {
        "pass": len(missing_inputs) == 0,
        "detail": {"missing": missing_inputs, "prev_run": prev_run_id},
    }

    # Set partition: every bg in exactly one subgroup
    seen: Dict[str, str] = {}
    duplicate: List[str] = []
    for sg in subgroups:
        sid = sg.get("subgroup_id")
        for m in (sg.get("member_bg_ids") or []):
            if m in seen:
                duplicate.append(f"{m}:in_{seen[m]}_and_{sid}")
            else:
                seen[m] = sid
    missing = sorted(plate_keys - set(seen.keys()))
    extra = sorted(set(seen.keys()) - plate_keys)
    inv["bg_set_partitioned_by_subgroups"] = {
        "pass": not duplicate and not missing and not extra,
        "detail": {"duplicate": duplicate[:5], "missing": missing[:5], "extra": extra[:5]},
    }

    # generation_chain is a subset of member_bg_ids, no dup, starts with base_bg_id
    chain_violations: List[str] = []
    for sg in subgroups:
        sid = sg.get("subgroup_id")
        members = set(sg.get("member_bg_ids") or [])
        chain = list(sg.get("generation_chain") or [])
        if len(chain) != len(set(chain)):
            chain_violations.append(f"{sid}:duplicate_in_chain")
        for c in chain:
            if c not in members:
                chain_violations.append(f"{sid}:chain_outside_members:{c}")
                break
        base = sg.get("base_bg_id")
        if base and base not in members:
            chain_violations.append(f"{sid}:base_outside_members:{base}")
        if chain and base and chain[0] != base:
            chain_violations.append(f"{sid}:chain_does_not_start_with_base")
    inv["generation_chain_is_member_subset_and_acyclic"] = {
        "pass": len(chain_violations) == 0,
        "detail": {"violating": chain_violations[:5]},
    }

    # base_bg_id present per subgroup
    no_base = [sg.get("subgroup_id") for sg in subgroups if not sg.get("base_bg_id")]
    inv["every_subgroup_has_base"] = {
        "pass": len(no_base) == 0,
        "detail": {"missing_base": no_base[:5]},
    }

    inv["no_human_decision_field"] = {
        "pass": True,
        "detail": "no banned keys at this stage",
    }
    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }
    inv["no_image_api_call"] = {
        "pass": not image_import_seen,
        "detail": "no openai images / gemini_image_client / fal import",
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


def _render_w8h_html(run_meta: dict, continuity: dict, report: dict, run_dir: Path) -> None:
    def esc(x):
        return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    subgroups = continuity.get("image_continuity_subgroups") or []
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td class=\"{'pass' if v['pass'] else 'fail'}\">{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    sg_rows = ""
    for sg in subgroups:
        members = ", ".join(esc(m) for m in (sg.get("member_bg_ids") or []))
        chain = " -> ".join(esc(c) for c in (sg.get("generation_chain") or []))
        sg_rows += (
            f"<tr><td>{esc(sg.get('subgroup_id'))}</td>"
            f"<td>{esc(sg.get('base_bg_id'))}</td>"
            f"<td>{members}</td><td>{chain}</td>"
            f"<td>{esc(sg.get('confidence'))}</td>"
            f"<td>{esc(sg.get('reason_brief'))}</td></tr>"
        )
    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice W8h {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em}}
table{{border-collapse:collapse}} td,th{{border:1px solid #ccc;padding:4px 8px}}
.pass{{color:#080}} .fail{{color:#b00}}</style></head>
<body>
<h1>W8h — image_continuity_subgroups {esc(run_meta.get('run_id'))}</h1>
<p>stage: <b>{esc(run_meta.get('stage'))}</b> | run_status: <b>{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}
| derived_from: {esc(run_meta.get('derived_from'))}
| model: <b>{esc(run_meta.get('model_used'))}</b></p>
<h2>Invariants</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Subgroups</h2>
<table><tr><th>subgroup_id</th><th>base_bg_id</th><th>members</th><th>generation_chain</th><th>conf</th><th>reason</th></tr>{sg_rows}</table>
</body></html>"""
    (run_dir / "index.html").write_text(html)


def _w8h_main(args, run_dir: Path, run_id: str) -> int:
    global _DB_WRITE_COUNT
    prev_run_dir = Path(args.derive_image_continuity_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir

    artifacts = _load_w7_artifacts(prev_run_dir)
    missing = list(artifacts.get("_missing", []))
    outputs: List[str] = []
    failed: List[str] = []
    run_status = "succeeded"
    exit_code = 0
    model_used = None

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W8H_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model_used": model_used,
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed,
    }
    if missing:
        failed.append("w7_inputs_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    w7_plan = artifacts["plan"]

    # Load W3 adapter + source bundle via chain
    try:
        adapter_plan = _load_w3_adapter_from_chain(prev_run_dir)
        resolved = _resolve_source_bundle_via_derived_from_chain(prev_run_dir)
        source_bundle = resolved["source_bundle"]
    except FileNotFoundError as exc:
        failed.append("w3_adapter_or_source_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        run_meta["error"] = str(exc)[:300]
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    llm_input = _build_continuity_llm_input(w7_plan, source_bundle, adapter_plan)

    if args.generate:
        _load_backend_env()
        try:
            continuity = _generate_image_continuity_via_llm(llm_input, model=args.model)
            model_used = args.model
        except Exception as exc:
            continuity = _build_placeholder_continuity(w7_plan, adapter_plan)
            failed.append("llm_call_failed")
            run_meta["error"] = str(exc)[:400]
    else:
        continuity = _build_placeholder_continuity(w7_plan, adapter_plan)

    (run_dir / "image_continuity_subgroups.json").write_text(
        json.dumps(continuity, ensure_ascii=False, indent=2)
    )
    outputs.append("image_continuity_subgroups.json")

    report = _build_continuity_compatibility_report(
        continuity=continuity, w7_plan=w7_plan,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_import_seen=_check_image_imports_present(),
        prev_run_id=prev_run_dir.name,
        missing_inputs=missing,
    )
    (run_dir / "image_continuity_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("image_continuity_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed:
            failed.append(name)
    if failed:
        run_status = "validation_failed"
        exit_code = 1

    run_meta["model_used"] = model_used
    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed
    run_meta["outputs"] = outputs
    _render_w8h_html(run_meta, continuity, report, run_dir)
    outputs.append("index.html")
    run_meta["outputs"] = outputs
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    _maybe_print_imports(args)
    return exit_code


# ─────────────────────────────────────────────────────────────────────────────
# W9 — W8 continuity-aware policy (uses W8h subgroups)
# ─────────────────────────────────────────────────────────────────────────────

def _load_continuity_subgroups(run_dir: Path) -> dict:
    p = run_dir / "image_continuity_subgroups.json"
    if not p.exists():
        raise FileNotFoundError(f"image_continuity_subgroups.json missing at {run_dir}")
    return json.loads(p.read_text())


def _build_bg_to_subgroup_index(continuity: dict) -> Dict[str, dict]:
    """bg_id -> {subgroup_id, base_bg_id, generation_chain, position_in_chain}."""
    out: Dict[str, dict] = {}
    for sg in (continuity.get("image_continuity_subgroups") or []):
        chain = list(sg.get("generation_chain") or [])
        for i, m in enumerate(sg.get("member_bg_ids") or []):
            out[m] = {
                "subgroup_id": sg.get("subgroup_id"),
                "base_bg_id": sg.get("base_bg_id"),
                "generation_chain": chain,
                "position_in_chain": chain.index(m) if m in chain else None,
            }
    return out


# ─────────────────────────────────────────────────────────────────────────────
# W11 — director_set_brief (read-only floor-plan context + LLM reconciliation)
# Trust hierarchy: hard structural (W3 adapter, source_bundle, applies_to_shots)
# > strong soft (floor_plan_prompt JSON numbered_elements/camera_recommendations)
# > weak visual (floor_plan_render PNG path/exists, no bytes read).
# LLM emits group_set_briefs + per_bg_set_selections + final_plate_prompt_candidates.
# ─────────────────────────────────────────────────────────────────────────────

W11_STAGE = "w11_director_set_brief"
W11_IMAGE_BACKEND = W6_IMAGE_BACKEND  # carry-only; W11 makes no image API call


def _load_production_floor_plan_context(project_id: str, episode_id: str) -> dict:
    """Read production floor_plan_prompt + floor_plan_render checkpoint
    manifests for the given project/episode. Read-only, structural carry only.

    Returns:
        {
            "fp_prompt_status": "ok" | "missing" | "parse_failed",
            "fp_render_status": "ok" | "missing" | "parse_failed",
            "floor_plans": { fp_id: {
                "fp_id", "group_id", "depends_on_fp",
                "key_elements", "numbered_elements", "camera_recommendations",
                "t2i_prompt", "applied_shots",
                "png_path", "png_exists",
            } },
            "fp_prompt_path", "fp_render_path",
        }
    No PNG bytes are read. ``png_exists`` is the only PNG-side fact carried.
    """
    base = _REPO_ROOT / "projects" / project_id / "checkpoints" / "episodes" / episode_id
    fp_prompt_path = base / "floor_plan_prompt" / "manifest.json"
    fp_render_path = base / "floor_plan_render" / "manifest.json"
    out: Dict[str, Any] = {
        "fp_prompt_path": str(fp_prompt_path),
        "fp_render_path": str(fp_render_path),
        "fp_prompt_status": "missing",
        "fp_render_status": "missing",
        "floor_plans": {},
    }
    fp_prompts: Dict[str, dict] = {}
    if fp_prompt_path.exists():
        try:
            mp = json.loads(fp_prompt_path.read_text())
            out["fp_prompt_status"] = "ok"
            fp_prompts = ((mp.get("data") or {}).get("floor_plans") or {})
        except Exception as exc:
            out["fp_prompt_status"] = "parse_failed"
            out["fp_prompt_error"] = str(exc)[:200]

    fp_renders: Dict[str, dict] = {}
    if fp_render_path.exists():
        try:
            mr = json.loads(fp_render_path.read_text())
            out["fp_render_status"] = "ok"
            fp_renders = ((mr.get("data") or {}).get("floor_plans") or {})
        except Exception as exc:
            out["fp_render_status"] = "parse_failed"
            out["fp_render_error"] = str(exc)[:200]

    floor_plans: Dict[str, dict] = {}
    for fp_id, fp in fp_prompts.items():
        if not isinstance(fp, dict):
            continue
        render_entry = fp_renders.get(fp_id) or {}
        png_path = render_entry.get("png_path") or ""
        png_exists = bool(png_path) and Path(png_path).exists()
        floor_plans[fp_id] = {
            "fp_id": fp_id,
            "group_id": fp.get("group_id"),
            "depends_on_fp": list(fp.get("depends_on_fp") or []),
            "key_elements": list(fp.get("key_elements") or []),
            "numbered_elements": list(fp.get("numbered_elements") or []),
            "camera_recommendations": list(fp.get("camera_recommendations") or []),
            "t2i_prompt": fp.get("t2i_prompt") or "",
            "applied_shots": list(fp.get("applied_shots") or []),
            "png_path": png_path,
            "png_exists": png_exists,
        }
    out["floor_plans"] = floor_plans
    return out


def _build_bg_to_fp_index(adapter_plan: dict) -> Dict[str, str]:
    """experiment adapter background_catalog[bg_id].depends_on_fp[0] → fp_id.
    Empty list / missing entry → empty string (no fp mapping)."""
    catalog = adapter_plan.get("background_catalog") or {}
    out: Dict[str, str] = {}
    for bg_id, entry in catalog.items():
        deps = (entry or {}).get("depends_on_fp") or []
        out[bg_id] = deps[0] if deps else ""
    return out


def _index_production_camera_recs_by_bg(fp_context: dict) -> Dict[str, dict]:
    """fp.camera_recommendations[].bg_id → recommendation entry. Production
    bg_id may not match experiment bg_id; exact-match lookup only, no semantic
    interpretation of bg_id strings."""
    out: Dict[str, dict] = {}
    for fp_id, fp in (fp_context.get("floor_plans") or {}).items():
        for cr in fp.get("camera_recommendations") or []:
            if not isinstance(cr, dict):
                continue
            bg_id = cr.get("bg_id") or ""
            if not bg_id:
                continue
            if bg_id in out:
                # production fp validator already exact-set; defensive first-wins
                continue
            out[bg_id] = {**cr, "_source_fp_id": fp_id}
    return out


LLM_W11_DIRECTOR_SET_BRIEF_SCHEMA = {
    "type": "object",
    "required": ["group_set_briefs", "per_bg_set_selections",
                 "final_plate_prompt_candidates"],
    "properties": {
        "group_set_briefs": {
            "type": "object",
            "additionalProperties": {
                "type": "object",
                "required": [
                    "group_id", "fp_blocks",
                    "spatial_zones", "openings_and_transitions",
                    "persistent_fixtures", "state_transition_zones",
                    "camera_axes", "default_avoidance",
                    "conflicts_and_assumptions", "director_set_summary",
                ],
                "properties": {
                    "group_id": {"type": "string"},
                    "fp_blocks": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "required": ["fp_id", "png_available",
                                         "production_camera_bg_ids"],
                            "properties": {
                                "fp_id": {"type": "string"},
                                "png_available": {"type": "boolean"},
                                "production_camera_bg_ids": {
                                    "type": "array", "items": {"type": "string"}
                                },
                            },
                        },
                    },
                    "spatial_zones": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "required": ["zone_id", "label_source_derived",
                                         "supporting_numbered_elements"],
                            "properties": {
                                "zone_id": {"type": "string"},
                                "label_source_derived": {"type": "string"},
                                "supporting_numbered_elements": {
                                    "type": "array",
                                    "items": {"type": "integer"},
                                },
                            },
                        },
                    },
                    "openings_and_transitions": {
                        "type": "array", "items": {"type": "string"}
                    },
                    "persistent_fixtures": {
                        "type": "array", "items": {"type": "string"}
                    },
                    "state_transition_zones": {
                        "type": "array", "items": {"type": "string"}
                    },
                    "camera_axes": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "required": ["axis_key", "from_zone", "to_zone", "reason"],
                            "properties": {
                                "axis_key": {"type": "string"},
                                "from_zone": {"type": "string"},
                                "to_zone": {"type": "string"},
                                "reason": {"type": "string"},
                            },
                        },
                    },
                    "default_avoidance": {"type": "string"},
                    "conflicts_and_assumptions": {
                        "type": "array", "items": {"type": "string"}
                    },
                    "director_set_summary": {"type": "string"},
                },
            },
        },
        "per_bg_set_selections": {
            "type": "object",
            "additionalProperties": {
                "type": "object",
                "required": [
                    "bg_id", "group_id", "fp_id",
                    "applies_to_shots", "relevant_zone_ids",
                    "relevant_numbered_elements",
                    "matched_production_camera",
                    "camera_axis_used", "continuity_intent",
                    "state_delta", "reconciliation_notes",
                ],
                "properties": {
                    "bg_id": {"type": "string"},
                    "group_id": {"type": "string"},
                    "fp_id": {"type": "string"},
                    "applies_to_shots": {
                        "type": "array", "items": {"type": "string"}
                    },
                    "relevant_zone_ids": {
                        "type": "array", "items": {"type": "string"}
                    },
                    "relevant_numbered_elements": {
                        "type": "array", "items": {"type": "integer"}
                    },
                    "matched_production_camera": {
                        "anyOf": [
                            {"type": "null"},
                            {
                                "type": "object",
                                "required": ["production_bg_id", "camera_position",
                                             "camera_height", "lens_hint"],
                                "properties": {
                                    "production_bg_id": {"type": "string"},
                                    "camera_position": {"type": "string"},
                                    "camera_height": {"type": "string"},
                                    "lens_hint": {"type": "string"},
                                    "framing_notes": {"type": "string"},
                                },
                            },
                        ],
                    },
                    "camera_axis_used": {"type": "string"},
                    "continuity_intent": {"type": "string"},
                    "state_delta": {"type": "string"},
                    "reconciliation_notes": {"type": "string"},
                },
            },
        },
        "final_plate_prompt_candidates": {
            "type": "object",
            "additionalProperties": {
                "type": "object",
                "required": ["bg_id", "final_plate_prompt_text",
                             "source_grounded_anchors_used",
                             "set_brief_grounded"],
                "properties": {
                    "bg_id": {"type": "string"},
                    "final_plate_prompt_text": {"type": "string"},
                    "source_grounded_anchors_used": {
                        "type": "array", "items": {"type": "string"}
                    },
                    "set_brief_grounded": {"type": "boolean"},
                },
            },
        },
    },
}


def _build_w11_llm_input(w7_plan: dict, adapter_plan: dict,
                         source_context: Dict[str, dict],
                         fp_context: dict) -> dict:
    """Compact view fed to W11 LLM. Generic carry only; no semantic interpretation
    of strings. PNG bytes are not included (only path + exists)."""
    plates = w7_plan.get("plate_prompts") or {}
    groups = w7_plan.get("groups") or []
    catalog = adapter_plan.get("background_catalog") or {}
    bg_to_fp = _build_bg_to_fp_index(adapter_plan)
    bg_to_group: Dict[str, str] = {}
    for g in groups:
        gid = g.get("group_id")
        for m in (g.get("member_bg_ids") or []):
            bg_to_group[m] = gid

    fp_by_id = fp_context.get("floor_plans") or {}
    # Compact fp view (per fp_id). PNG bytes NOT included.
    fp_views: Dict[str, dict] = {}
    for fp_id, fp in fp_by_id.items():
        fp_views[fp_id] = {
            "fp_id": fp_id,
            "key_elements": fp.get("key_elements"),
            "numbered_elements": fp.get("numbered_elements"),
            "camera_recommendations": fp.get("camera_recommendations"),
            "t2i_prompt": fp.get("t2i_prompt"),
            "applied_shots": fp.get("applied_shots"),
            "png_path": fp.get("png_path"),
            "png_exists": fp.get("png_exists"),
            "render_status": "available" if fp.get("png_exists") else "no_png",
        }

    # Per-bg view: experiment bg_id + adapter facts + W7 plate text + bg→fp lookup.
    bg_views: List[dict] = []
    for bg_id, p in plates.items():
        catalog_entry = catalog.get(bg_id) or {}
        bg_views.append({
            "bg_id": bg_id,
            "group_id": bg_to_group.get(bg_id),
            "fp_id_via_adapter": bg_to_fp.get(bg_id, ""),
            "plate_prompt_text": p.get("plate_prompt_text"),
            "prompt_role": p.get("prompt_role"),
            "reference_strength": p.get("reference_strength"),
            "render_mode": p.get("render_mode"),
            "applies_to_shots": p.get("applies_to_shots"),
            "depends_on_bg": catalog_entry.get("depends_on_bg"),
            "sub_location_label": catalog_entry.get("sub_location_label"),
            "state_label_raw": catalog_entry.get("state_label_raw"),
            "loc_id": catalog_entry.get("loc_id"),
            "space_key": catalog_entry.get("space_key"),
            "time_phase": catalog_entry.get("time_phase"),
            "state_class": catalog_entry.get("state_class"),
        })

    # Group views: group→fp blocks; per-bg side already carries fp_id_via_adapter.
    group_views: List[dict] = []
    for g in groups:
        gid = g.get("group_id")
        member_bgs = list(g.get("member_bg_ids") or [])
        fp_block_ids: List[str] = []
        for m in member_bgs:
            fid = bg_to_fp.get(m, "")
            if fid and fid not in fp_block_ids:
                fp_block_ids.append(fid)
        group_views.append({
            "group_id": gid,
            "base_bg_id": g.get("base_bg_id"),
            "member_bg_ids": member_bgs,
            "loc_id": g.get("loc_id"),
            "space_key": g.get("space_key"),
            "fp_block_ids": fp_block_ids,
            "source_context": source_context.get(gid),
        })

    return {
        "background_entries": bg_views,
        "groups": group_views,
        "floor_plans": fp_views,
        "json_schema": LLM_W11_DIRECTOR_SET_BRIEF_SCHEMA,
    }


W11_SYSTEM_PROMPT = (
    "You are a film-production set designer/director. Your task: produce a "
    "director_set_brief that reconciles three layers of evidence about a "
    "shooting location: "
    "(A) HARD structural evidence — adapter background_catalog (bg_id, "
    "depends_on_bg, sub_location_label, state_label_raw, loc_id, space_key), "
    "selected_shots, applies_to_shots. These are authoritative IDs and facts. "
    "(B) STRONG soft evidence — production floor_plan_prompt JSON: "
    "numbered_elements (category/label/position_hint), camera_recommendations "
    "(production bg_ids; may NOT equal experiment bg_ids), t2i_prompt, "
    "key_elements. Treat the JSON as the layout source of truth at "
    "set-design level. "
    "(C) WEAK visual evidence — floor_plan_render PNG path/exists. The image "
    "is a non-pixel-accurate diagram. Do NOT cite individual marker pixel "
    "positions from the PNG; you only know it exists. "
    "For every group, emit a group_set_brief. group_set_briefs[group_id]."
    "fp_blocks must list PRODUCTION fp_id strings (the keys of the input "
    "floor_plans object) that the group maps to — choose them based on "
    "loc_id / sub_location_label / applied_shots overlap. Within fp_blocks, "
    "set production_camera_bg_ids by copying the bg_id values from those "
    "production fp's camera_recommendations entries; set png_available from "
    "the same fp's png_exists boolean. Then enumerate spatial_zones using the "
    "floor-plan numbered_elements (cite supporting_numbered_elements by "
    "integer; numbers must come from the production fp's numbered_elements), "
    "describe openings_and_transitions (doors, sliding windows, stair access), "
    "persistent_fixtures, state_transition_zones (where state changes will be "
    "staged), camera_axes (from_zone -> to_zone with a short reason). "
    "default_avoidance must be a generic non-upgrade clause: 'preserve the "
    "scale, material condition, age/finish level, and visual class already "
    "stated or implied by the source. Do not upscale, modernize, sanitize, or "
    "add adjacent areas/objects beyond the per-bg prompt.' "
    "For every per-bg selection: set the per_bg fp_id field to the EXPERIMENT "
    "adapter value (the bg view's fp_id_via_adapter) verbatim — NEVER a "
    "production fp_id from the floor_plans context. The PRODUCTION fp_id may "
    "differ from the EXPERIMENT adapter fp_id. The group_set_briefs[group_id]."
    "fp_blocks list the PRODUCTION fp_ids that map to this group (LLM-chosen "
    "from the floor_plans context based on loc_id / sub_location / applied_shots). "
    "When picking relevant_numbered_elements, use integer numbers from the "
    "numbered_elements of the PRODUCTION fp_id(s) listed in that group's "
    "fp_blocks. If no production fp matches the group, leave "
    "relevant_numbered_elements empty and explain in reconciliation_notes. "
    "Choose camera_axis_used (axis_key from the group's camera_axes); set "
    "matched_production_camera = the production camera_recommendation entry "
    "only when its bg_id EXACTLY equals the experiment bg_id, else null and "
    "instead derive camera_axis_used + a fresh reconciliation_notes line. Do "
    "not invent a production_bg_id. "
    "For every bg, emit a final_plate_prompt_candidates entry whose "
    "final_plate_prompt_text reads as a single-paragraph background-only "
    "plate-prompt: camera framing + zone + relevant fixtures, source-derived "
    "wording, no character entities, no scene_specific story facts that aren't "
    "in applies_to_shots / source_context. set_brief_grounded must be true. "
    "Surface conflicts_and_assumptions whenever floor-plan JSON wording "
    "disagrees with source_context shots (e.g. plate text from W7 names a "
    "different room than the fp numbered_elements suggest). HARD wins. "
    "Output strict JSON matching the provided json_schema; no extra keys, "
    "no commentary."
)


def _generate_director_set_brief_via_llm(llm_input: dict, *, model: str,
                                         retry_once: bool = True) -> dict:
    """--generate path. Single LLM call (one retry on JSON validation failure)."""
    import litellm

    user_prompt = json.dumps(llm_input, ensure_ascii=False)
    routed = (
        model
        if model.startswith("gemini/") or not model.lower().startswith("gemini")
        else f"gemini/{model}"
    )
    if not (os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")):
        raise RuntimeError("missing GEMINI_API_KEY / GOOGLE_API_KEY env var")
    last_exc: Optional[Exception] = None
    attempts = 2 if retry_once else 1
    for _ in range(attempts):
        try:
            resp = litellm.completion(
                model=routed,
                messages=[
                    {"role": "system", "content": W11_SYSTEM_PROMPT},
                    {"role": "user", "content": user_prompt},
                ],
                response_format={"type": "json_object"},
            )
            raw_text = resp.choices[0].message.content
            decoder = json.JSONDecoder()
            stripped = raw_text.lstrip()
            parsed, _ = decoder.raw_decode(stripped)
            import jsonschema
            jsonschema.validate(parsed, LLM_W11_DIRECTOR_SET_BRIEF_SCHEMA)
            return parsed
        except Exception as exc:
            last_exc = exc
            continue
    raise RuntimeError(f"w11_llm_failed_after_retry: {last_exc!s}"[:400])


def _build_placeholder_director_set_brief(w7_plan: dict, adapter_plan: dict,
                                          fp_context: dict) -> dict:
    """Dry-run — emit shell entries (placeholders) for every bg/group. status
    encoded in the candidate's source_grounded_anchors_used=['placeholder_dry_run']."""
    plates = w7_plan.get("plate_prompts") or {}
    groups = w7_plan.get("groups") or []
    bg_to_fp = _build_bg_to_fp_index(adapter_plan)
    fp_by_id = fp_context.get("floor_plans") or {}

    group_set_briefs: Dict[str, dict] = {}
    for g in groups:
        gid = g.get("group_id")
        member_bgs = list(g.get("member_bg_ids") or [])
        fp_block_ids: List[str] = []
        for m in member_bgs:
            fid = bg_to_fp.get(m, "")
            if fid and fid not in fp_block_ids:
                fp_block_ids.append(fid)
        fp_blocks = []
        for fid in fp_block_ids:
            fp = fp_by_id.get(fid) or {}
            production_camera_bg_ids = [
                cr.get("bg_id", "") for cr in (fp.get("camera_recommendations") or [])
                if cr.get("bg_id")
            ]
            fp_blocks.append({
                "fp_id": fid,
                "png_available": bool(fp.get("png_exists")),
                "production_camera_bg_ids": production_camera_bg_ids,
            })
        group_set_briefs[gid] = {
            "group_id": gid,
            "fp_blocks": fp_blocks,
            "spatial_zones": [],
            "openings_and_transitions": [],
            "persistent_fixtures": [],
            "state_transition_zones": [],
            "camera_axes": [],
            "default_avoidance": "placeholder_dry_run",
            "conflicts_and_assumptions": [],
            "director_set_summary": "placeholder_dry_run",
        }

    per_bg_set_selections: Dict[str, dict] = {}
    final_plate_prompt_candidates: Dict[str, dict] = {}
    bg_to_group: Dict[str, str] = {}
    for g in groups:
        gid = g.get("group_id")
        for m in (g.get("member_bg_ids") or []):
            bg_to_group[m] = gid
    for bg_id, p in plates.items():
        per_bg_set_selections[bg_id] = {
            "bg_id": bg_id,
            "group_id": bg_to_group.get(bg_id, ""),
            "fp_id": bg_to_fp.get(bg_id, ""),
            "applies_to_shots": list(p.get("applies_to_shots") or []),
            "relevant_zone_ids": [],
            "relevant_numbered_elements": [],
            "matched_production_camera": None,
            "camera_axis_used": "",
            "continuity_intent": "placeholder_dry_run",
            "state_delta": "placeholder_dry_run",
            "reconciliation_notes": "placeholder_dry_run",
        }
        final_plate_prompt_candidates[bg_id] = {
            "bg_id": bg_id,
            "final_plate_prompt_text": p.get("plate_prompt_text") or "",
            "source_grounded_anchors_used": ["placeholder_dry_run"],
            "set_brief_grounded": False,
        }
    return {
        "group_set_briefs": group_set_briefs,
        "per_bg_set_selections": per_bg_set_selections,
        "final_plate_prompt_candidates": final_plate_prompt_candidates,
    }


def _build_w11_compatibility_report(
    *, brief: dict, w7_plan: dict, adapter_plan: dict, fp_context: dict,
    production_diff_empty: bool, db_write_count: int,
    image_import_seen: bool, prev_run_id: str, missing_inputs: List[str],
    set_brief_status: str,
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}
    plate_keys = set((w7_plan.get("plate_prompts") or {}).keys())
    bg_to_fp = _build_bg_to_fp_index(adapter_plan)
    fp_by_id = fp_context.get("floor_plans") or {}

    inv["w7_inputs_present"] = {
        "pass": len(missing_inputs) == 0,
        "detail": {"missing": missing_inputs, "prev_run": prev_run_id},
    }
    inv["floor_plan_inputs_resolved"] = {
        "pass": fp_context.get("fp_prompt_status") == "ok",
        "detail": {
            "fp_prompt_status": fp_context.get("fp_prompt_status"),
            "fp_render_status": fp_context.get("fp_render_status"),
            "fp_count": len(fp_by_id),
        },
    }

    # per-bg relevant_numbered_elements ⊆ union of numbered_elements numbers
    # from the production fp_ids listed in group_set_briefs[group_id].fp_blocks.
    # Adapter fp_id (per_bg.fp_id) may not match any production fp_id, so the
    # subset is computed against the group's chosen production fp_blocks; this
    # is the LLM's explicit reconciliation surface.
    per_bg = brief.get("per_bg_set_selections") or {}
    group_briefs = brief.get("group_set_briefs") or {}

    def _allowed_numbers_for_group(gid: str) -> set:
        blocks = ((group_briefs.get(gid) or {}).get("fp_blocks") or [])
        return {
            int(e.get("number"))
            for blk in blocks
            for e in ((fp_by_id.get(blk.get("fp_id") or "") or {}).get("numbered_elements") or [])
            if isinstance(e.get("number"), int)
        }

    subset_violations: List[str] = []
    for bg_id, sel in per_bg.items():
        gid = sel.get("group_id") or ""
        picked = list(sel.get("relevant_numbered_elements") or [])
        if not picked:
            continue
        allowed = _allowed_numbers_for_group(gid)
        out_of_set = [n for n in picked if int(n) not in allowed]
        if out_of_set:
            subset_violations.append(f"{bg_id}:group={gid}:extra={out_of_set[:5]}")
    inv["per_bg_relevant_numbered_elements_subset_of_fp"] = {
        "pass": len(subset_violations) == 0,
        "detail": {"violating": subset_violations[:5]},
    }

    # per-bg.fp_id == adapter bg_to_fp[bg_id]
    map_violations: List[str] = []
    for bg_id, sel in per_bg.items():
        expected_fp = bg_to_fp.get(bg_id, "")
        got_fp = sel.get("fp_id") or ""
        if expected_fp and got_fp and expected_fp != got_fp:
            map_violations.append(f"{bg_id}:expected={expected_fp}:got={got_fp}")
    inv["bg_to_fp_mapping_consistent_with_adapter"] = {
        "pass": len(map_violations) == 0,
        "detail": {"violating": map_violations[:5]},
    }

    # set partition: every plate bg has both a per_bg and a final_plate_prompt_candidates entry
    cands = brief.get("final_plate_prompt_candidates") or {}
    missing_per_bg = sorted(plate_keys - set(per_bg.keys()))
    missing_cands = sorted(plate_keys - set(cands.keys()))
    inv["every_plate_bg_has_brief_and_candidate"] = {
        "pass": not missing_per_bg and not missing_cands,
        "detail": {"missing_per_bg": missing_per_bg[:5],
                   "missing_candidate": missing_cands[:5]},
    }

    # all final candidates have set_brief_grounded == True (LLM path).
    # Dry-run placeholder path is exempt; this invariant only applies when the
    # LLM actually ran (set_brief_status='generated'). validation_failed fails it.
    not_grounded = [bg for bg, c in cands.items() if not c.get("set_brief_grounded")]
    if set_brief_status == "placeholder_dry_run":
        grounded_pass = True
    elif set_brief_status == "generated":
        grounded_pass = len(not_grounded) == 0
    else:
        grounded_pass = False
    inv["final_plate_prompt_candidates_set_brief_grounded"] = {
        "pass": grounded_pass,
        "detail": {"status": set_brief_status, "not_grounded": not_grounded[:5]},
    }

    # banned human-decision keys
    BANNED = {"needs_user", "manual_review", "decision", "awaiting_human"}

    def _has_banned(obj: Any) -> bool:
        if isinstance(obj, dict):
            if any(k in BANNED for k in obj.keys()):
                return True
            return any(_has_banned(v) for v in obj.values())
        if isinstance(obj, list):
            return any(_has_banned(x) for x in obj)
        return False

    inv["no_human_decision_field"] = {
        "pass": not _has_banned(brief),
        "detail": "no banned keys at this stage",
    }

    inv["production_diff_zero"] = {
        "pass": bool(production_diff_empty),
        "detail": "git diff backend/app backend/alembic empty",
    }
    inv["db_write_zero"] = {
        "pass": db_write_count == 0,
        "detail": f"writes={db_write_count}",
    }
    inv["no_image_api_call"] = {
        "pass": not image_import_seen,
        "detail": "no openai images / gemini_image_client / fal import",
    }
    inv["image_generation_count_zero"] = {
        "pass": True,
        "detail": "W11 stage emits zero PNGs by design",
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


def _render_w11_html(run_meta: dict, brief: dict, report: dict,
                     fp_context: dict, run_dir: Path) -> None:
    def esc(x):
        return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td class=\"{'pass' if v['pass'] else 'fail'}\">{'PASS' if v['pass'] else 'FAIL'}</td>"
        f"<td>{esc(json.dumps(v.get('detail'), ensure_ascii=False))[:300]}</td></tr>"
        for k, v in inv.items()
    )

    # group_set_briefs section
    group_rows = ""
    for gid, g in (brief.get("group_set_briefs") or {}).items():
        zones = ", ".join(
            f"{esc(z.get('zone_id'))} (#{','.join(str(n) for n in z.get('supporting_numbered_elements') or [])})"
            for z in (g.get("spatial_zones") or [])
        )
        axes = " | ".join(
            f"{esc(ax.get('axis_key'))}: {esc(ax.get('from_zone'))} → {esc(ax.get('to_zone'))}"
            for ax in (g.get("camera_axes") or [])
        )
        conflicts = "<br>".join(esc(c) for c in (g.get("conflicts_and_assumptions") or []))
        fp_block_summary = ", ".join(
            f"{esc(fb.get('fp_id'))}(png={'Y' if fb.get('png_available') else 'N'},"
            f"prod_bgs={len(fb.get('production_camera_bg_ids') or [])})"
            for fb in (g.get("fp_blocks") or [])
        )
        group_rows += (
            f"<tr><td>{esc(gid)}</td>"
            f"<td>{fp_block_summary}</td>"
            f"<td>{zones}</td>"
            f"<td>{axes}</td>"
            f"<td>{conflicts}</td>"
            f"<td>{esc(g.get('director_set_summary'))[:500]}</td></tr>"
        )

    # per-bg selections section
    bg_rows = ""
    for bg_id, sel in (brief.get("per_bg_set_selections") or {}).items():
        nums = ",".join(str(n) for n in (sel.get("relevant_numbered_elements") or []))
        mc = sel.get("matched_production_camera")
        mc_str = "null" if mc is None else esc(mc.get("production_bg_id", ""))
        bg_rows += (
            f"<tr><td>{esc(bg_id)}</td>"
            f"<td>{esc(sel.get('group_id'))}</td>"
            f"<td>{esc(sel.get('fp_id'))}</td>"
            f"<td>{nums}</td>"
            f"<td>{esc(sel.get('camera_axis_used'))}</td>"
            f"<td>{mc_str}</td>"
            f"<td>{esc(sel.get('reconciliation_notes'))[:300]}</td></tr>"
        )

    # final candidates section
    cand_rows = ""
    for bg_id, c in (brief.get("final_plate_prompt_candidates") or {}).items():
        anchors = ", ".join(esc(a) for a in (c.get("source_grounded_anchors_used") or []))
        cand_rows += (
            f"<tr><td>{esc(bg_id)}</td>"
            f"<td>{'Y' if c.get('set_brief_grounded') else 'N'}</td>"
            f"<td>{anchors}</td>"
            f"<td><pre>{esc(c.get('final_plate_prompt_text') or '')[:1500]}</pre></td></tr>"
        )

    # fp context section
    fp_rows = ""
    for fp_id, fp in (fp_context.get("floor_plans") or {}).items():
        prod_bgs = ",".join(
            cr.get("bg_id", "") for cr in (fp.get("camera_recommendations") or [])
            if cr.get("bg_id")
        )
        fp_rows += (
            f"<tr><td>{esc(fp_id)}</td>"
            f"<td>{len(fp.get('numbered_elements') or [])}</td>"
            f"<td>{prod_bgs}</td>"
            f"<td>{'Y' if fp.get('png_exists') else 'N'}</td>"
            f"<td>{esc(fp.get('png_path') or '')}</td></tr>"
        )

    html = f"""<!doctype html><html><head><meta charset=\"utf-8\">
<title>background_pipeline_slice W11 {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em}}
table{{border-collapse:collapse;margin:0.5em 0}} td,th{{border:1px solid #ccc;padding:4px 8px;vertical-align:top}}
.pass{{color:#080}} .fail{{color:#b00}}
pre{{white-space:pre-wrap;font-size:0.85em;max-width:60ch}}
section{{margin:1.5em 0}}</style></head>
<body>
<h1>W11 — director_set_brief {esc(run_meta.get('run_id'))}</h1>
<p>stage: <b>{esc(run_meta.get('stage'))}</b> | run_status: <b>{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}
| derived_from: {esc(run_meta.get('derived_from'))}
| model: <b>{esc(run_meta.get('model_used'))}</b>
| image_generation_count: <b>{esc(run_meta.get('image_generation_count'))}</b>
| image_generation_backend: <b>{esc(run_meta.get('image_generation_backend'))}</b></p>

<section><h2>1. Trust hierarchy</h2>
<ol>
<li><b>HARD structural</b> — W3 adapter background_catalog, source_bundle.selected_shots, applies_to_shots, loc/space IDs. Exact id matching only.</li>
<li><b>STRONG soft</b> — production floor_plan_prompt JSON: numbered_elements (category/label/position_hint), camera_recommendations (production bg_id), key_elements, t2i_prompt.</li>
<li><b>WEAK visual</b> — floor_plan_render PNG path/exists only. No bytes read.</li>
</ol></section>

<section><h2>2. Invariants</h2>
<table><tr><th>invariant</th><th>status</th><th>detail</th></tr>{inv_rows}</table></section>

<section><h2>3. Director group briefs</h2>
<table><tr><th>group_id</th><th>fp_blocks</th><th>spatial_zones</th><th>camera_axes</th><th>conflicts</th><th>director_set_summary</th></tr>{group_rows}</table></section>

<section><h2>4. Per-bg set selections</h2>
<table><tr><th>bg_id</th><th>group</th><th>fp_id</th><th>relevant_#</th><th>camera_axis</th><th>matched_prod_camera</th><th>reconciliation_notes</th></tr>{bg_rows}</table></section>

<section><h2>5. Final plate prompt candidates</h2>
<table><tr><th>bg_id</th><th>set_brief_grounded</th><th>anchors_used</th><th>final_plate_prompt_text</th></tr>{cand_rows}</table></section>

<section><h2>6. Floor-plan context resolution</h2>
<p>fp_prompt_path: {esc(fp_context.get('fp_prompt_path'))}<br>
fp_render_path: {esc(fp_context.get('fp_render_path'))}<br>
fp_prompt_status: <b>{esc(fp_context.get('fp_prompt_status'))}</b> |
fp_render_status: <b>{esc(fp_context.get('fp_render_status'))}</b></p>
<table><tr><th>fp_id</th><th>numbered#</th><th>production_camera_bg_ids</th><th>png_exists</th><th>png_path</th></tr>{fp_rows}</table></section>

<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


def _w11_main(args, run_dir: Path, run_id: str) -> int:
    global _DB_WRITE_COUNT
    prev_run_dir = Path(args.derive_director_set_brief_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir

    artifacts = _load_w7_artifacts(prev_run_dir)
    missing = list(artifacts.get("_missing", []))
    outputs: List[str] = []
    failed: List[str] = []
    run_status = "succeeded"
    exit_code = 0
    model_used = None
    set_brief_status = "placeholder_dry_run"

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W11_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model": args.model if args.generate else None,
        "model_used": model_used,
        "image_generation_count": 0,
        "image_generation_backend": W11_IMAGE_BACKEND,
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed,
    }

    if missing:
        failed.append("w7_inputs_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    w7_plan = artifacts["plan"]

    # Resolve W3 adapter + source_bundle via derived_from chain.
    try:
        adapter_plan = _load_w3_adapter_from_chain(prev_run_dir)
        resolved = _resolve_source_bundle_via_derived_from_chain(prev_run_dir)
        source_bundle = resolved["source_bundle"]
    except FileNotFoundError as exc:
        failed.append("w3_adapter_or_source_missing")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        run_meta["error"] = str(exc)[:300]
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code

    # Build per-group source context (reused W7c helper).
    # W6 plan needed for group source_context shot_keys; W7 plan carries groups[].shot_keys.
    source_context = _build_per_group_source_context(w7_plan, source_bundle)

    # Read production floor_plan_prompt + floor_plan_render manifests.
    project_id = source_bundle.get("project_id") or ""
    episode_id = source_bundle.get("episode_id") or ""
    if not project_id or not episode_id:
        failed.append("project_or_episode_id_missing_in_source_bundle")
        run_status = "validation_failed"
        exit_code = 1
        run_meta["run_status"] = run_status
        run_meta["exit_code"] = exit_code
        run_meta["failed_invariants"] = failed
        (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
        return exit_code
    fp_context = _load_production_floor_plan_context(project_id, episode_id)

    llm_input = _build_w11_llm_input(w7_plan, adapter_plan, source_context, fp_context)

    if args.generate:
        _load_backend_env()
        try:
            brief = _generate_director_set_brief_via_llm(llm_input, model=args.model)
            model_used = args.model
            set_brief_status = "generated"
            # mark candidates grounded after successful LLM (LLM also sets it,
            # but we defensively enforce true on the experiment side).
            for c in (brief.get("final_plate_prompt_candidates") or {}).values():
                c["set_brief_grounded"] = True
        except Exception as exc:
            brief = _build_placeholder_director_set_brief(w7_plan, adapter_plan, fp_context)
            set_brief_status = "validation_failed"
            failed.append("llm_call_failed")
            run_meta["error"] = str(exc)[:400]
    else:
        brief = _build_placeholder_director_set_brief(w7_plan, adapter_plan, fp_context)

    (run_dir / "director_set_brief_plan.json").write_text(
        json.dumps(brief, ensure_ascii=False, indent=2)
    )
    outputs.append("director_set_brief_plan.json")

    report = _build_w11_compatibility_report(
        brief=brief, w7_plan=w7_plan, adapter_plan=adapter_plan,
        fp_context=fp_context,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=_DB_WRITE_COUNT,
        image_import_seen=_check_image_imports_present(),
        prev_run_id=prev_run_dir.name,
        missing_inputs=missing,
        set_brief_status=set_brief_status,
    )
    (run_dir / "director_set_brief_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    outputs.append("director_set_brief_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed:
            failed.append(name)
    if failed:
        run_status = "validation_failed"
        exit_code = 1

    run_meta["model_used"] = model_used
    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed
    run_meta["outputs"] = outputs
    run_meta["set_brief_status"] = set_brief_status
    _render_w11_html(run_meta, brief, report, fp_context, run_dir)
    outputs.append("index.html")
    run_meta["outputs"] = outputs
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    _maybe_print_imports(args)
    return exit_code


if __name__ == "__main__":
    sys.exit(main())