#!/usr/bin/env python3 """ Read-only attached-reference trace extractor. - No code modifications, no DB writes. - Outputs: TSV per episode + location catalog JSON per episode + console summary. """ import json import os import re import subprocess import sys from collections import Counter, defaultdict from pathlib import Path OUT = Path("/Users/manta/Documents/Projects/TheRoad-I1/scripts_output/bg_ref_trace") OUT.mkdir(parents=True, exist_ok=True) DSN_ENV = {"PGPASSWORD": "theroad_dev_2026"} EPISODES = { "new": { "project_id": "6cb862d9-590c-4dce-86e6-d10c2977db19", "episode_id": "08ad2cd3-3e96-4d84-808f-869ee628473c", "label": "Task 14 fresh full E2E (post reference-necessity)", }, "c10p1": { "project_id": "76212b49-bf96-45fb-8c56-cd8d8ae02dda", "episode_id": "f44339e6-1bd1-4d10-8a8f-2e30e5a1d36d", "label": "C10 Phase 1 fresh full E2E", }, } SEP = "\x1f" # ASCII Unit Separator — safe vs tabs/newlines in text fields def psql(query: str) -> str: env = os.environ.copy() env.update(DSN_ENV) res = subprocess.run( ["psql", "-h", "localhost", "-U", "theroad", "-d", "theroad", "-t", "-A", "-F", SEP, "-R", "\x1e\n", "-c", query], capture_output=True, text=True, env=env, check=True, ) # Use record separator \x1e + \n so multi-line text fields don't break parsing. return res.stdout def psql_rows(query: str): """Yield split rows using \x1e record separator (filters trailing empty).""" raw = psql(query) # rows separated by \x1e + newline; last row may have no trailing sep for rec in raw.split("\x1e\n"): rec = rec.rstrip("\n").rstrip("\x1e") if not rec.strip(): continue yield rec.split(SEP) def fetch_locations(project_id: str): """Return {short_id: {id, name, space_profile}} for locations.""" q = ( "SELECT id, short_id, name, metadata_json FROM entity_canon " f"WHERE project_id='{project_id}' AND entity_type='location' " "ORDER BY short_id" ) out = {} for parts in psql_rows(q): cid, sid, name, meta = parts[0], parts[1], parts[2], parts[3] try: md = json.loads(meta) if meta else {} except Exception: md = {} out[sid] = {"canon_id": cid, "short_id": sid, "name": name, "metadata": md} return out def fetch_image_assets(project_id: str): """Return list of all image assets for project (id, asset_type, entity_id, file_path, still_id).""" q = ( "SELECT id, asset_type, COALESCE(entity_id,''), COALESCE(file_path,''), COALESCE(still_id,''), COALESCE(variant_label,''), COALESCE(theme_label,'') " f"FROM image_asset WHERE project_id='{project_id}'" ) rows = [] for parts in psql_rows(q): rows.append({ "id": parts[0], "asset_type": parts[1], "entity_id": parts[2], "file_path": parts[3], "still_id": parts[4], "variant_label": parts[5] if len(parts) > 5 else "", "theme_label": parts[6] if len(parts) > 6 else "", }) return rows LOC_BG_RE = re.compile(r"(L\d+)B\d+\.png", re.IGNORECASE) LOC_FP_RE = re.compile(r"fp_(l\d+|[a-z_]+)_", re.IGNORECASE) def derive_loc_from_chain_bg_path(path: str) -> str: m = LOC_BG_RE.search(path) return m.group(1) if m else "" def fetch_selected_shots(episode_id: str): """Return list of selected scene_still rows with key fields.""" q = ( "SELECT id, scene_index, shot_index, scene_summary, shot_description, beat_title, " "COALESCE(visible_entities_json,'[]'), COALESCE(t2i_variations_json,'[]'), COALESCE(dependent_scene_id,'') " f"FROM scene_still WHERE episode_id='{episode_id}' AND is_selected=true " "ORDER BY scene_index, shot_index" ) rows = [] for parts in psql_rows(q): if len(parts) < 9: print(f"WARN unexpected col count {len(parts)} -> {parts[0] if parts else 'EMPTY'}", file=sys.stderr) continue rid, sidx, shidx, ssum, sdesc, btitle, ve_json, t2i_json, dep_id = parts[:9] try: ve = json.loads(ve_json) if ve_json else [] except Exception: ve = [] try: t2i = json.loads(t2i_json) if t2i_json else [] except Exception: t2i = [] rows.append({ "still_id": rid, "scene_index": int(sidx) if sidx else 0, "shot_index": int(shidx) if shidx else 0, "scene_summary": ssum, "shot_description": sdesc, "beat_title": btitle, "visible_entities": ve, "t2i_variations": t2i, "dependent_scene_id": dep_id or None, }) return rows def fetch_scene_image_assets(project_id: str): """Return {still_id: scene image_asset row (with reference_image_ids list + prompt_used)}""" q = ( "SELECT still_id, file_path, COALESCE(reference_image_ids,'[]'), id, COALESCE(prompt_used,'') " f"FROM image_asset WHERE project_id='{project_id}' AND asset_type='scene' " "AND still_id IS NOT NULL AND still_id != ''" ) out = {} for parts in psql_rows(q): sid, fp, refs_json, iid, prompt_used = parts[0], parts[1], parts[2], parts[3], parts[4] if len(parts) > 4 else "" try: refs = json.loads(refs_json) if refs_json else [] except Exception: refs = [] cur = out.get(sid) new_entry = {"file_path": fp, "reference_image_ids": refs, "asset_id": iid, "prompt_used": prompt_used} if cur is None or len(refs) > len(cur["reference_image_ids"]): out[sid] = new_entry return out # Pattern: [LXXBYY: ...] inline bg descriptor in the scene prompt PROMPT_BG_PAT = re.compile(r"\[(L\d+B\d+):", re.IGNORECASE) PROMPT_FP_PAT = re.compile(r"fp_[a-z0-9_]+", re.IGNORECASE) PROMPT_PREV_PAT = re.compile(r"PREVIOUS SHOT", re.IGNORECASE) PROMPT_BG_CHAIN_PAT = re.compile(r"pre-rendered BACKGROUND chain reference", re.IGNORECASE) def parse_prompt_attachments(prompt_used: str): """Extract bg codes (e.g. ['L05B03', 'L08B01']), fp codes, prev_shot flag from prompt text.""" if not prompt_used: return {"bg_codes": [], "fp_codes": [], "prev_shot": False, "bg_chain_block": False, "has_any_ref": False} bg = PROMPT_BG_PAT.findall(prompt_used) or [] fp = PROMPT_FP_PAT.findall(prompt_used) or [] return { "bg_codes": list(dict.fromkeys(bg)), # preserve order, dedupe "fp_codes": list(dict.fromkeys(fp)), "prev_shot": bool(PROMPT_PREV_PAT.search(prompt_used)), "bg_chain_block": bool(PROMPT_BG_CHAIN_PAT.search(prompt_used)), "has_any_ref": bool(re.search(r"Reference image \d", prompt_used)) or PROMPT_BG_CHAIN_PAT.search(prompt_used) is not None or PROMPT_PREV_PAT.search(prompt_used) is not None, } def build_asset_index(assets): """Index image_assets by id for quick join lookup.""" return {a["id"]: a for a in assets} def classify_ref(asset, loc_id_by_canon_id, ec_short_by_canon_id, ec_type_by_canon_id): """Given an image_asset row referenced by a scene, return (kind, label). kind ∈ {character, prop, chain_bg, floor_plan, reference_other, unknown} label = short identifier (entity short_id + suffix).""" atype = asset["asset_type"] eid = asset["entity_id"] fp = asset["file_path"] short = ec_short_by_canon_id.get(eid, "") etype = ec_type_by_canon_id.get(eid, "") if atype == "chain_bg": loc_short = short or loc_id_by_canon_id.get(eid, "") bg_code = derive_loc_from_chain_bg_path(fp) # filename may already start with location short_id bgfile = os.path.basename(fp).replace(".png", "") return ("chain_bg", f"{loc_short}/{bgfile}") if atype == "floor_plan": loc_short = short or loc_id_by_canon_id.get(eid, "") bgfile = os.path.basename(fp).replace(".png", "") return ("floor_plan", f"{loc_short}/{bgfile}") if atype == "reference": # character or prop or outlook reference if etype == "character": return ("character", short) if etype == "prop": return ("prop", short) if etype == "outlook": return ("outlook", short) if etype == "location": # base location reference (rare) return ("chain_bg", short) return ("reference_other", f"{short}:{etype}") return ("unknown", f"{atype}:{short}") def extract_visible_locations(visible_entities): """Return short_id list of location entities visible (from scene_director visible_entities_json).""" out = [] for v in visible_entities: sid = (v.get("short_id") or "").strip() if sid.startswith("L") and re.match(r"^L\d+$", sid): out.append(sid) return out def extract_visible_all(visible_entities): """All visible entity short_ids.""" out = [] for v in visible_entities: sid = v.get("short_id") or "" if sid: out.append(sid) return out def summarize_ref_usage_per_var(t2i_variations): """Compact reference_phrase_kinds per variation as 'v1:char,bg | v2:character'.""" parts = [] for v in t2i_variations: label = v.get("variant_label", "v?") kinds = v.get("reference_phrase_kinds") or [] parts.append(f"{label}:{','.join(kinds) if kinds else '-'}") return " | ".join(parts) def short_text(s: str, n: int = 120) -> str: s = (s or "").replace("\n", " ").replace("\t", " ").strip() if len(s) <= n: return s return s[:n] + "…" def safe_tsv(s) -> str: if s is None: return "" return str(s).replace("\t", " ").replace("\n", " ").replace("\r", " ") def main(): summary_per_ep = {} for alias, info in EPISODES.items(): project_id = info["project_id"] episode_id = info["episode_id"] print(f"\n=== Processing {alias}: project={project_id} episode={episode_id} ===", file=sys.stderr) locations = fetch_locations(project_id) assets = fetch_image_assets(project_id) asset_idx = build_asset_index(assets) ec_short_by_canon_id = {} ec_type_by_canon_id = {} loc_id_by_canon_id = {} # build canon_id -> short_id maps from entity_canon q_ec = ( "SELECT id, COALESCE(short_id,''), entity_type FROM entity_canon " f"WHERE project_id='{project_id}'" ) for p in psql_rows(q_ec): ec_short_by_canon_id[p[0]] = p[1] ec_type_by_canon_id[p[0]] = p[2] if p[2] == "location": loc_id_by_canon_id[p[0]] = p[1] shots = fetch_selected_shots(episode_id) scene_image_by_still = fetch_scene_image_assets(project_id) # Build still_id -> shot lookup for dependent_scene labels shot_by_still = {s["still_id"]: s for s in shots} # dependent could also point at non-selected stills; fetch those too dep_ids = {s["dependent_scene_id"] for s in shots if s["dependent_scene_id"]} missing = dep_ids - set(shot_by_still.keys()) dep_extra = {} if missing: ids_in = "','".join(missing) q = ( "SELECT id, scene_index, shot_index, COALESCE(scene_summary,'') FROM scene_still " f"WHERE id IN ('{ids_in}')" ) for p in psql_rows(q): dep_extra[p[0]] = { "scene_index": int(p[1]) if p[1] else 0, "shot_index": int(p[2]) if p[2] else 0, "scene_summary": p[3] if len(p) > 3 else "", } # also fetch their visible_entities for PREV_SHOT_DIFF_LOC q = ( "SELECT id, COALESCE(visible_entities_json,'[]') FROM scene_still " f"WHERE id IN ('{ids_in}')" ) for p in psql_rows(q): try: dep_extra[p[0]]["visible_entities"] = json.loads(p[1]) except Exception: dep_extra[p[0]]["visible_entities"] = [] # ====== build over_split groups (chain_bg variants per location) ====== chain_by_loc_short = defaultdict(list) for a in assets: if a["asset_type"] == "chain_bg": short = ec_short_by_canon_id.get(a["entity_id"], "") if short: bg = os.path.basename(a["file_path"]).replace(".png", "") chain_by_loc_short[short].append({"bg_code": bg, "image_id": a["id"]}) fp_by_loc_short = defaultdict(list) for a in assets: if a["asset_type"] == "floor_plan": short = ec_short_by_canon_id.get(a["entity_id"], "") if short: fp_by_loc_short[short].append({ "fp_name": os.path.basename(a["file_path"]).replace(".png", ""), "image_id": a["id"], }) catalog_loc = [] over_split = [] unused_locs = [] for sid, loc in sorted(locations.items()): chains = chain_by_loc_short.get(sid, []) fps = fp_by_loc_short.get(sid, []) entry = { "short_id": sid, "name": loc["name"], "space_profile": loc["metadata"].get("location", {}).get("space_profile") if isinstance(loc["metadata"], dict) else None, "chain_bg_count": len(chains), "chain_bg_variants": [c["bg_code"] for c in chains], "floor_plan_count": len(fps), "floor_plan_variants": [f["fp_name"] for f in fps], } catalog_loc.append(entry) if len(chains) >= 3: over_split.append({ "base_location_id": sid, "name": loc["name"], "variant_count": len(chains), "variants": [c["bg_code"] for c in chains], }) if len(chains) == 0 and len(fps) == 0: unused_locs.append({"short_id": sid, "name": loc["name"]}) # Save world_guide raw notes (text guardrails only, since no structured catalog in DB) # also note: there is no parent_location_id / chain field in DB schema -> isolated_variants n/a structurally catalog_path = OUT / f"{alias}_location_catalog.json" catalog = { "project_id": project_id, "episode_id": episode_id, "label": info["label"], "total_locations": len(locations), "total_chain_bg_assets": sum(len(v) for v in chain_by_loc_short.values()), "total_floor_plan_assets": sum(len(v) for v in fp_by_loc_short.values()), "over_split_groups": over_split, "unused_locations": unused_locs, "raw_locations": catalog_loc, "note_isolated_variants": ( "DB schema has no parent_location_id / chain key on chain_bg assets — chain_bg variants " "are always linked to a single LXX location entity via image_asset.entity_id (UUID). " "'Over-split' here means many chain_bg image variants per LXX location (Cat A symptom). " "There is NO LXXBYY level entity_canon row — the BXX suffix exists only in file_path. " "Therefore 'isolated_variants' (Cat B as originally defined) does not apply structurally; " "Cat B must be reframed as 'multiple chain_bg variants of the same LXX never used together' " "(see TSV mismatch_flag_candidates for MULTI_BG / OVER_SPLIT_HOST)." ), } with catalog_path.open("w", encoding="utf-8") as f: json.dump(catalog, f, ensure_ascii=False, indent=2) print(f"wrote {catalog_path}", file=sys.stderr) # ====== TSV trace ====== tsv_path = OUT / f"{alias}_61shots_trace.tsv" header = [ "scene_index", "shot_index", "still_id", "scene_summary", "shot_description_short", "visible_location_claim", "scene_director_visible_locations", "scene_director_visible_entities_all", "attached_ref_count", "attached_char_refs", "attached_prop_refs", "attached_bg_refs_from_prompt", "attached_floor_plan_refs", "prompt_has_bg_chain_block", "prompt_has_prev_shot_block", "ref_usage_per_variation", "dependent_scene_id", "dependent_scene_label", "scene_image_path", "mismatch_flag_candidates", ] per_shot_summary = [] flag_counter = Counter() per_cat_examples = defaultdict(list) with tsv_path.open("w", encoding="utf-8") as f: f.write("\t".join(header) + "\n") for s in shots: sid = s["still_id"] scene_img = scene_image_by_still.get(sid, {}) refs_ids = scene_img.get("reference_image_ids", []) prompt_used = scene_img.get("prompt_used", "") prompt_attach = parse_prompt_attachments(prompt_used) # join refs to image_asset — these are character/prop refs only (chain_bg/fp are NOT in reference_image_ids per investigation) joined = [asset_idx[rid] for rid in refs_ids if rid in asset_idx] # classify kinds = [] char_refs = [] prop_refs = [] other_refs = [] for a in joined: kind, label = classify_ref(a, loc_id_by_canon_id, ec_short_by_canon_id, ec_type_by_canon_id) if kind == "character": char_refs.append(label) elif kind == "prop": prop_refs.append(label) elif kind == "outlook": other_refs.append(f"outlook:{label}") elif kind == "chain_bg": # rare — base location reference other_refs.append(f"bg-ref:{label}") elif kind == "floor_plan": other_refs.append(f"fp-ref:{label}") else: other_refs.append(f"{kind}:{label}") kinds.append(kind) # bg_refs / fp_refs come from prompt parsing (SOT for chain_bg attachment in this pipeline) bg_refs = prompt_attach["bg_codes"] # e.g. ['L05B03'] fp_refs = prompt_attach["fp_codes"] # visible locations from scene_director vis_locs = extract_visible_locations(s["visible_entities"]) vis_all = extract_visible_all(s["visible_entities"]) # visible_location_claim: shot_description+scene_summary text text_blob = (s["shot_description"] or "") + " | " + (s["scene_summary"] or "") claim = text_blob # raw, no truncation per CLAUDE.md absolute rule # mismatch flags flags = [] # bg_refs are LXXBYY codes from prompt parsing — strip BYY to get host location attached_bg_locs = set() bg_code_to_host = re.compile(r"^(L\d+)B\d+$", re.IGNORECASE) for b in bg_refs: m = bg_code_to_host.match(b) if m: attached_bg_locs.add(m.group(1).upper()) attached_fp_locs = set() # floor_plans never attached to scenes in either E2E # LOCATION_MISMATCH = visible_locs && attached_bg_locs, but disjoint if vis_locs and attached_bg_locs and not (set(vis_locs) & attached_bg_locs): flags.append("LOCATION_MISMATCH") # BG_MISSING = visible_locs exists but no bg refs from prompt AND no PREVIOUS_SHOT reframing # (i.e. scene has spatial location but no chain_bg/floor_plan/prev_shot anchor at all) if vis_locs and not bg_refs and not fp_refs and not prompt_attach["prev_shot"]: flags.append("BG_MISSING") # MULTI_BG = >=2 different host locations in attached bg if len(attached_bg_locs) >= 2: flags.append("MULTI_BG") # PREV_SHOT_DIFF_LOC dep_id = s["dependent_scene_id"] dep_label = "" if dep_id: dep_obj = shot_by_still.get(dep_id) or dep_extra.get(dep_id) if dep_obj: dep_vis = extract_visible_locations(dep_obj.get("visible_entities", [])) dep_label = f"S{dep_obj['scene_index']}_Shot{dep_obj['shot_index']} ({short_text(dep_obj['scene_summary'], 60)})" if dep_vis and vis_locs: diff = (set(dep_vis) ^ set(vis_locs)) if diff: flags.append("PREV_SHOT_DIFF_LOC") else: dep_label = f"" # OVER_SPLIT_HOST = attached_bg loc is in over_split set over_split_ids = {o["base_location_id"] for o in over_split} if attached_bg_locs & over_split_ids: flags.append("OVER_SPLIT_HOST") for fl in flags: flag_counter[fl] += 1 # categorize for examples # Cat A (over-split): attached_bg in over_split host if "OVER_SPLIT_HOST" in flags: per_cat_examples["A"].append(( s["scene_index"], s["shot_index"], sid, f"L={sorted(attached_bg_locs)} variant={','.join(bg_refs)} (host has {len([o for o in over_split if o['base_location_id'] in attached_bg_locs])} over-split locations)" )) # Cat B (missing shared structure proxy): MULTI_BG of same-base or same-loc multi-variant if "MULTI_BG" in flags: per_cat_examples["B"].append((s["scene_index"], s["shot_index"], sid, f"attached_bg_locs={sorted(attached_bg_locs)} bg_refs={bg_refs}")) # Cat C (wrong bg): LOCATION_MISMATCH or BG_MISSING if "LOCATION_MISMATCH" in flags: per_cat_examples["C"].append((s["scene_index"], s["shot_index"], sid, f"visible={vis_locs} attached_bg={sorted(attached_bg_locs)} attached_fp={sorted(attached_fp_locs)}")) if "BG_MISSING" in flags: per_cat_examples["C"].append((s["scene_index"], s["shot_index"], sid, f"visible={vis_locs} attached_bg=∅ attached_fp=∅ (refs={[x for x in kinds]})")) # Cat D (prev_shot pollution) if "PREV_SHOT_DIFF_LOC" in flags: per_cat_examples["D"].append((s["scene_index"], s["shot_index"], sid, f"this={vis_locs} dep_label={dep_label}")) row = [ s["scene_index"], s["shot_index"], sid, safe_tsv(s["scene_summary"]), safe_tsv(short_text(s["shot_description"], 120)), safe_tsv(claim), json.dumps(vis_locs, ensure_ascii=False), json.dumps(vis_all, ensure_ascii=False), len(joined), json.dumps(char_refs, ensure_ascii=False), json.dumps(prop_refs, ensure_ascii=False), json.dumps(bg_refs, ensure_ascii=False), json.dumps(fp_refs, ensure_ascii=False), "1" if prompt_attach["bg_chain_block"] else "0", "1" if prompt_attach["prev_shot"] else "0", safe_tsv(summarize_ref_usage_per_var(s["t2i_variations"])), dep_id or "", safe_tsv(dep_label), scene_img.get("file_path", ""), ";".join(flags), ] f.write("\t".join(str(x) for x in row) + "\n") per_shot_summary.append({ "scene_index": s["scene_index"], "shot_index": s["shot_index"], "still_id": sid, "flags": flags, "vis_locs": vis_locs, "bg_refs": bg_refs, "fp_refs": fp_refs, "char_refs": char_refs, "prop_refs": prop_refs, }) print(f"wrote {tsv_path} ({len(shots)} rows)", file=sys.stderr) summary_per_ep[alias] = { "info": info, "shot_count": len(shots), "flag_counter": dict(flag_counter), "per_cat_examples": dict(per_cat_examples), "over_split": over_split, "unused_locs": unused_locs, "per_shot_summary": per_shot_summary, "loc_count": len(locations), "chain_bg_total": sum(len(v) for v in chain_by_loc_short.values()), "fp_total": sum(len(v) for v in fp_by_loc_short.values()), } # ====== final console summary ====== print("\n" + "=" * 80) print("BG/REF TRACE — FINAL SUMMARY") print("=" * 80) print("\n[Output files]") for alias in EPISODES: print(f" /Users/manta/Documents/Projects/TheRoad-I1/scripts_output/bg_ref_trace/{alias}_61shots_trace.tsv") print(f" /Users/manta/Documents/Projects/TheRoad-I1/scripts_output/bg_ref_trace/{alias}_location_catalog.json") print("\n[Mismatch flag counts per episode]") flag_names = ["LOCATION_MISMATCH", "BG_MISSING", "MULTI_BG", "PREV_SHOT_DIFF_LOC", "OVER_SPLIT_HOST"] print(f" {'flag':<22} | {'new':>5} | {'c10p1':>6}") print(f" {'-'*22} + {'-'*5} + {'-'*6}") for fl in flag_names: a = summary_per_ep["new"]["flag_counter"].get(fl, 0) b = summary_per_ep["c10p1"]["flag_counter"].get(fl, 0) print(f" {fl:<22} | {a:>5} | {b:>6}") print("\n[Catalog overview]") for alias, d in summary_per_ep.items(): print(f" {alias}: locations={d['loc_count']}, chain_bg_assets={d['chain_bg_total']}, fp_assets={d['fp_total']}, over_split_groups={len(d['over_split'])} (>=3), unused_locs={len(d['unused_locs'])}") print("\n[Cat A — over-split chain_bg (top examples)]") for alias, d in summary_per_ep.items(): print(f" -- {alias}") for o in sorted(d["over_split"], key=lambda x: -x["variant_count"])[:5]: print(f" {o['base_location_id']} ({o['name']}) -> {o['variant_count']} variants: {o['variants']}") print("\n[Per-category example shots (3 each per episode)]") cat_titles = { "A": "Cat A — attached_bg refs an over-split host location", "B": "Cat B — MULTI_BG (>=2 different locations attached)", "C": "Cat C — LOCATION_MISMATCH or BG_MISSING", "D": "Cat D — PREV_SHOT_DIFF_LOC (dependent scene differs in visible location)", } for cat in "ABCD": print(f"\n {cat_titles[cat]}") for alias, d in summary_per_ep.items(): ex = d["per_cat_examples"].get(cat, [])[:3] print(f" -- {alias} (total {len(d['per_cat_examples'].get(cat, []))} flagged)") for sidx, shidx, still, note in ex: print(f" S{sidx}_Shot{shidx} still={still[:8]} :: {note}") print("\n[Cross-episode quick contrast]") new = summary_per_ep["new"]; c10 = summary_per_ep["c10p1"] print(f" flag delta (new - c10p1):") for fl in flag_names: a = new["flag_counter"].get(fl, 0); b = c10["flag_counter"].get(fl, 0) print(f" {fl:<22} : {a-b:+d} (new={a}, c10p1={b})") print("\n[CRITICAL data-gap finding]") print(" - scene image_asset.reference_image_ids 는 character/prop reference assets 만 포함.") print(" chain_bg / floor_plan 은 reference_image_ids 에 절대 들어가지 않음 (전체 73 ref 중 0).") print(" - prompt_used 텍스트에 \"pre-rendered BACKGROUND chain reference\" 블록이 new 33/61, c10p1 27/61 등장,") print(" 하지만 [LXXBYY: ...] 인라인 코드가 같이 박힌 케이스는 new 3/61, c10p1 2/61 뿐.") print(" → 대부분의 chain_bg attach 는 \"Reference image 1\" 라벨로만 들어가서 어떤 LXXBYY 가 첨부됐는지") print(" DB 어디에도 persist 되지 않음 (t2i_guide=0, generation_trace=0, reference_image_ids=무관).") print(" - floor_plan asset 은 두 episode 모두 scene prompt 에 0건 인용 (생성됐지만 사용 0).") print(" → 결과: Cat A/B/C 의 실제 분포는 text-trace 가능한 shot 만 부분 진단 가능.") print(" BG_MISSING 다수는 \"진짜 ref 없음\" 이 아니라 \"image-only attach, 텍스트엔 코드 누락\" 일 가능성.") print("\n[W1 decision candidates]") print(" 1) chain_bg attachment 추적 가능성 결정 — image_asset 에 attach_origin/chain_bg_id 컬럼 추가하여") print(" scene render 시 어떤 LXXBYY 가 multimodal 로 들어갔는지 persist 할지 (현 SOT 부재).") print(" 2) over-split host location LXX 의 N개 chain_bg variant 를 base + state-overlay 로 합칠지.") print(" - new L05(옥탑방 내부, 8 variants), c10p1 L04(옥탑방 내부, 9 variants) 최대.") print(" - entity_canon 차원에 LXX→LXXBYY 관계 추가 vs file_path-only 패턴 유지.") print(" 3) floor_plan 무사용 정책 — 두 E2E 22 fp asset 생성·0 attach. (a) fallback 으로 활용 (b) 생성 자체 skip.") print(" 4) BG_MISSING (visible loc 있는데 bg/fp/prev_shot 0) — new 31/61, c10p1 24/61.") print(" → 코드 게이트 추가하여 chain_bg 없는 location 의 shot 은 char-only 로 명시할지, 아니면 자동 fp fallback 할지.") print(" 5) prompt 내 [LXXBYY: ...] 마크업 일관성 — 일부는 박히고 대부분 안 박힘.") print(" producer-side fix 로 모든 bg-chain attached shot 의 prompt 에 강제 inline 코드 삽입할지.") if __name__ == "__main__": main()