#!/usr/bin/env python3 """Rooftop Source Grounding Experiment — standalone (plan v1 APPROVED 2026-05-23). plan_v1: scripts_output/rooftop_source_grounding/plan_v1.md Codex Q1 fix: GeminiTextClient 직접 import 금지 (log_llm_call → DB write). script-local urllib. production pipeline 0 수정, DB write 0, dry-run 기본. """ from __future__ import annotations import argparse import base64 # noqa: F401 (호환) import html import json import re import sys import time import uuid from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Optional _REPO_ROOT = Path(__file__).resolve().parents[2] _BACKEND_ROOT = _REPO_ROOT / "backend" if str(_BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(_BACKEND_ROOT)) def _load_backend_env() -> None: import os as _os # noqa: PLC0415 env_path = _BACKEND_ROOT / ".env" if not env_path.exists(): return for raw in env_path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() if (value.startswith('"') and value.endswith('"')) or ( value.startswith("'") and value.endswith("'") ): value = value[1:-1] if key and key not in _os.environ: _os.environ[key] = value _load_backend_env() from app.core.database import SessionLocal # noqa: E402 from app.models.catalog import ProjectRegistry # noqa: E402 from app.models.project import Episode # noqa: E402 # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19" EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c" L05_SHORT_ID = "L05" DEFAULT_OUTPUT_DIR = Path("scripts_output/rooftop_source_grounding") DEFAULT_BASE_RUN_PARENT = Path("scripts_output/rooftop_spatial_bg_experiment") GEMINI_FLASH_DEFAULT = "gemini-3.5-flash" # Deterministic keyword categories (사용자 명시 + Codex Q2 권고). KEYWORD_CATEGORIES: dict[str, list[str]] = { "sub_space": [ "옥탑방", "옥상", "거실", "안방", "침실", "방 안", "방안", "욕실", "화장실", "주방", "부엌", "현관", ], "fixture": [ "식탁", "싱크대", "침대", "욕조", "변기", "거울", "창문", "커튼", "벽면", "바닥", "천장", "문", "계단", ], "state": [ "피", "시신", "핏자국", "정돈", "어지럽", "깨끗", "어두운", "빈", "표식", ], "condition_socio": [ "낡은", "좁은", "오래된", "작은", "옥탑", "다세대", "빌라", ], } KEYWORD_FIELD_HINTS: dict[str, str] = { "옥탑방": "space.location_root", "옥상": "space.adjacent", "거실": "sub_space.main_room", "안방": "sub_space.bedroom", "침실": "sub_space.bedroom", "방 안": "sub_space.bedroom_or_main", "방안": "sub_space.bedroom_or_main", "욕실": "sub_space.bathroom", "화장실": "sub_space.bathroom", "주방": "sub_space.kitchen", "부엌": "sub_space.kitchen", "현관": "sub_space.entry", "식탁": "furniture.dining_table", "싱크대": "furniture.sink", "침대": "furniture.bed", "욕조": "fixture.bathtub", "변기": "fixture.toilet", "거울": "fixture.mirror", "창문": "doors_windows.window", "커튼": "fixture.curtain", "벽면": "material.wall", "바닥": "material.floor", "천장": "material.ceiling", "문": "doors_windows.door", "계단": "structure.stairs", "피": "state.corpse_marks", "시신": "state.corpse_marks", "핏자국": "state.corpse_marks", "정돈": "state.cleaned", "어지럽": "state.vandalized", "깨끗": "state.cleaned", "어두운": "state.night", "빈": "state.empty", "표식": "state.vandalized_or_mark", "낡은": "condition_age.old", "좁은": "condition_age.cramped", "오래된": "condition_age.old", "작은": "condition_age.small", "옥탑": "space.rooftop", "다세대": "socioeconomic.lower_to_middle", "빌라": "socioeconomic.lower_to_middle", } # Codex Q3: luxury_cue candidate only — "옥탑방=저소득" truth 가 아님. LUXURY_CUE_CANDIDATES = [ "marble", "chandelier", "island kitchen", "open-plan kitchen", "floor-to-ceiling window", "walk-in closet", "hotel lighting", "designer loft", "large sectional sofa", "luxury bathroom", "high ceiling", "polished concrete", "wine fridge", "double-height", "panoramic view", "skyline view", ] SEVERITY_CANDIDATES = ["info", "suspicious", "likely_mismatch"] HTML_QUOTE_LIMIT = 140 # Codex 추가 권고 # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass class EvidenceRow: source: str # 'episode' | 'planning_doc' keyword: str category: str # sub_space | fixture | state | condition_socio candidate_field: str quote: str char_offset: int confidence: str # 'strong' | 'weak' (단순 — 매칭 자체가 weak/strong) notes: str = "" @dataclass class SourceBundle: episode_fulltext: str planning_doc_text: str missing_inputs: list[dict] = field(default_factory=list) episode_meta: dict = field(default_factory=dict) project_meta: dict = field(default_factory=dict) # --------------------------------------------------------------------------- # DB read-only # --------------------------------------------------------------------------- def load_source_bundle(session) -> SourceBundle: """episode fulltext + planning_doc_text 수집 + missing inputs 명시.""" bundle = SourceBundle(episode_fulltext="", planning_doc_text="") ep = session.query(Episode).filter(Episode.id == EPISODE_ID).one_or_none() if ep is None: bundle.missing_inputs.append({ "key": "episode_fulltext", "looked_at": f"episode.fulltext where id={EPISODE_ID}", "reason": "episode row 부재", }) else: bundle.episode_fulltext = (ep.fulltext or "").strip() bundle.episode_meta = { "id": ep.id, "episode_number": ep.episode_number, "title": ep.title, "source_filename": ep.source_filename, "source_path": ep.source_path, "fulltext_length": len(bundle.episode_fulltext), "summary_length": len(ep.summary or ""), } if not bundle.episode_fulltext: bundle.missing_inputs.append({ "key": "episode_fulltext", "looked_at": "episode.fulltext", "reason": "fulltext 비어 있음", }) proj = session.query(ProjectRegistry).filter(ProjectRegistry.id == PROJECT_ID).one_or_none() if proj is None: bundle.missing_inputs.append({ "key": "planning_doc_text", "looked_at": f"project_registry.planning_doc_text where id={PROJECT_ID}", "reason": "project row 부재", }) else: bundle.planning_doc_text = (proj.planning_doc_text or "").strip() bundle.project_meta = { "id": proj.id, "name": proj.name, "planning_doc_length": len(bundle.planning_doc_text), } if not bundle.planning_doc_text: bundle.missing_inputs.append({ "key": "planning_doc_text", "looked_at": "project_registry.planning_doc_text", "reason": "planning_doc_text 비어 있음", }) # world_summary / style_rules — project_settings row 없으면 모두 missing. from sqlalchemy import text as _sa_text # noqa: PLC0415 rows = session.execute(_sa_text( "SELECT world_summary, style_rules_json FROM project_settings WHERE project_id=:pid" ), {"pid": PROJECT_ID}).fetchall() if not rows: bundle.missing_inputs.append({ "key": "world_summary", "looked_at": "project_settings where project_id=" + PROJECT_ID, "reason": "project_settings row 부재", }) bundle.missing_inputs.append({ "key": "style_rules_json", "looked_at": "project_settings where project_id=" + PROJECT_ID, "reason": "project_settings row 부재", }) return bundle # --------------------------------------------------------------------------- # Deterministic extraction # --------------------------------------------------------------------------- def _extract_quote(text: str, idx: int, k_len: int, max_chars: int = 200) -> str: """idx 위치 기준 sentence-ish window. 마침표/줄바꿈 가장자리.""" start = idx end = idx + k_len # back: 최근 . 또는 \n 또는 시작 back = max(text.rfind(".", 0, start), text.rfind("\n", 0, start), text.rfind("?", 0, start), text.rfind("!", 0, start)) if back == -1: sentence_start = max(0, start - max_chars // 2) else: sentence_start = back + 1 # forward: 다음 . 또는 \n candidates = [ text.find(".", end), text.find("\n", end), text.find("?", end), text.find("!", end), ] candidates = [c for c in candidates if c != -1] sentence_end = min(candidates) + 1 if candidates else min(len(text), end + max_chars // 2) if sentence_end - sentence_start > max_chars: sentence_end = sentence_start + max_chars return text[sentence_start:sentence_end].strip() def extract_evidence(bundle: SourceBundle) -> list[EvidenceRow]: rows: list[EvidenceRow] = [] for source_label, body in ( ("episode", bundle.episode_fulltext), ("planning_doc", bundle.planning_doc_text), ): if not body: continue for category, kws in KEYWORD_CATEGORIES.items(): for kw in kws: start = 0 while True: idx = body.find(kw, start) if idx == -1: break quote = _extract_quote(body, idx, len(kw)) rows.append(EvidenceRow( source=source_label, keyword=kw, category=category, candidate_field=KEYWORD_FIELD_HINTS.get(kw, ""), quote=quote, char_offset=idx, confidence="strong" if category == "sub_space" else "weak", )) start = idx + len(kw) return rows # --------------------------------------------------------------------------- # Gemini strict-JSON wrapper (Codex Q1: script-local, NOT GeminiTextClient) # --------------------------------------------------------------------------- ROOFTOP_BIBLE_SCHEMA: dict = { "type": "object", "properties": { "sub_spaces": {"type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "evidence_quotes": {"type": "array", "items": {"type": "string"}}, "layout_notes": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["name", "evidence_quotes", "layout_notes", "confidence_band"], }}, "layout_relations": {"type": "array", "items": { "type": "object", "properties": { "from_sub_space": {"type": "string"}, "to_sub_space": {"type": "string"}, "relation": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["from_sub_space", "to_sub_space", "relation", "evidence", "confidence_band"], }}, "doors_windows": {"type": "array", "items": { "type": "object", "properties": { "kind": {"type": "string"}, "location": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["kind", "location", "evidence", "confidence_band"], }}, "kitchen": {"type": "object", "properties": { "present": {"type": "boolean"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["present", "evidence", "confidence_band"]}, "bathroom": {"type": "object", "properties": { "present": {"type": "boolean"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["present", "evidence", "confidence_band"]}, "furniture": {"type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "sub_space": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["name", "sub_space", "evidence", "confidence_band"], }}, "materials": {"type": "array", "items": { "type": "object", "properties": { "where": {"type": "string"}, "material": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["where", "material", "evidence", "confidence_band"], }}, "condition_age": {"type": "object", "properties": { "summary": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["summary", "evidence", "confidence_band"]}, "socioeconomic_tone": {"type": "object", "properties": { "summary": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["summary", "evidence", "confidence_band"]}, "scale_density": {"type": "object", "properties": { "summary": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["summary", "evidence", "confidence_band"]}, "state_variants": {"type": "array", "items": { "type": "object", "properties": { "state": {"type": "string"}, "evidence": {"type": "string"}, "confidence_band": {"type": "string"}, }, "required": ["state", "evidence", "confidence_band"], }}, "required_visual_cues": {"type": "array", "items": {"type": "string"}}, "forbidden_luxury_cues": {"type": "array", "items": {"type": "string"}}, "unknowns": {"type": "array", "items": {"type": "string"}}, "inference_limits": {"type": "array", "items": {"type": "string"}}, }, "required": [ "sub_spaces", "layout_relations", "doors_windows", "kitchen", "bathroom", "furniture", "materials", "condition_age", "socioeconomic_tone", "scale_density", "state_variants", "required_visual_cues", "forbidden_luxury_cues", "unknowns", "inference_limits", ], } # Codex Blocking (post-asset-copy): str.format() 가 literal {trusted, weak, unknown} 을 placeholder # 로 오인 → KeyError. helper 안에서 단순 replace 로 치환. literal brace 그대로 유지 OK. GEMINI_INSTRUCTION_TEMPLATE = ( "당신은 한국 시나리오 분석가입니다. 아래 episode 전문 + 기획안 전문을 evidence 기반으로 읽고, " "L05 '옥탑방 내부' (이 작품의 주거 공간) 의 source bible 을 strict JSON 으로 추출하세요.\n\n" "원칙:\n" "1. evidence quote 를 반드시 원문에서 직접 인용. 없으면 빈 문자열.\n" "2. 확실히 추론 가능한 것만. 불확실하면 unknowns 또는 inference_limits 로.\n" "3. 시각화 결정자가 아닙니다. forbidden_luxury_cues 는 source 와 명백히 어긋난 화려한 표현 후보만 (예: 샹들리에, 대리석, 호텔 조명).\n" "4. confidence_band ∈ {trusted, weak, unknown}. >=0.75=trusted, 0.5~0.75=weak, <0.5=unknown.\n" "5. metric 표기 (m, mm, 평) 금지 — qualitative summary 만.\n" "6. 가족 구성, 등장인물 외형, 사건 전개는 무시 (이 bible 은 '공간 자체' 만).\n\n" "입력 텍스트는 다음 두 블록으로 구분되어 있습니다:\n" "=== EPISODE 1 FULLTEXT ===\n<>\n\n" "=== PROJECT PLANNING DOC ===\n<>\n" ) def build_gemini_prompt(episode_text: str, planning_text: str) -> str: """Codex Blocking fix: str.format placeholder 충돌 회피. `{trusted, weak, unknown}` 같은 literal brace 가 instruction 안에 있어서 `.format(episode=..., planning=...)` 가 KeyError 로 터짐. <> 토큰 사용. """ return ( GEMINI_INSTRUCTION_TEMPLATE .replace("<>", episode_text) .replace("<>", planning_text) ) # 하위 호환 — 기존 test 의 GEMINI_INSTRUCTION 참조 유지. GEMINI_INSTRUCTION = GEMINI_INSTRUCTION_TEMPLATE def call_gemini_bible( *, episode_text: str, planning_text: str, gemini_model: str, cost_log_path: Optional[Path] = None, call_meta_log_path: Optional[Path] = None, max_attempts: int = 2, ) -> dict: """Codex Q1: script-local urllib wrapper, NOT GeminiTextClient (DB write 방지).""" import socket as _socket # noqa: PLC0415 import urllib.error as _urlerr # noqa: PLC0415 import urllib.request as _urlreq # noqa: PLC0415 import os as _os # noqa: PLC0415 from app.core.config import settings # noqa: PLC0415 # Codex Important 1: settings.gemini_api_key (단수) → env GEMINI_API_KEY fallback. # production gemini_key_pool 의 multi-key rotation 은 실험엔 불필요. api_key = (getattr(settings, "gemini_api_key", "") or "").strip() if not api_key: api_key = _os.environ.get("GEMINI_API_KEY", "").strip() if not api_key: return {"status": "failed", "error": "GEMINI_API_KEY 미설정"} prompt_body = build_gemini_prompt(episode_text, planning_text) body = { "contents": [{ "role": "user", "parts": [{"text": prompt_body}], }], "generationConfig": { "responseMimeType": "application/json", "responseSchema": ROOFTOP_BIBLE_SCHEMA, "temperature": 0.1, }, } url = ( f"https://generativelanguage.googleapis.com/v1beta/models/{gemini_model}:generateContent" f"?key={api_key}" ) headers = {"Content-Type": "application/json"} last_exc: Optional[Exception] = None for attempt in range(1, max_attempts + 1): t0 = time.perf_counter() try: req = _urlreq.Request(url, data=json.dumps(body).encode("utf-8"), headers=headers, method="POST") with _urlreq.urlopen(req, timeout=180) as resp: payload = json.loads(resp.read().decode("utf-8")) ms = int((time.perf_counter() - t0) * 1000) # parts[0].text → strict JSON string text = ( payload.get("candidates", [{}])[0] .get("content", {}).get("parts", [{}])[0].get("text", "") ) if not text: raise RuntimeError(f"empty Gemini response: {json.dumps(payload)[:200]}") bible = json.loads(text) # Codex Blocking 2: local jsonschema validate (responseSchema 만 의존 X). import jsonschema # noqa: PLC0415 try: jsonschema.validate(instance=bible, schema=ROOFTOP_BIBLE_SCHEMA) except jsonschema.ValidationError as ve: raise RuntimeError(f"jsonschema validation failed: {str(ve)[:200]}") from ve _append_jsonl_if(cost_log_path, { "stage": "gemini_bible", "model": gemini_model, "status": "ok", "latency_ms": ms, }) _append_jsonl_if(call_meta_log_path, { "kind": "gemini_call", "stage": "gemini_bible", "model": gemini_model, "prompt_chars": len(prompt_body), "latency_ms": ms, "status": "ok", }) return {"status": "ok", "bible": bible, "latency_ms": ms, "attempts": attempt} except (_urlerr.HTTPError, _urlerr.URLError, _socket.timeout, RuntimeError, json.JSONDecodeError) as exc: last_exc = exc ms = int((time.perf_counter() - t0) * 1000) _append_jsonl_if(cost_log_path, { "stage": "gemini_bible", "model": gemini_model, "status": "error", "latency_ms": ms, "error": str(exc)[:200], "attempt": attempt, }) if attempt >= max_attempts: return {"status": "failed", "error": str(exc)[:300], "attempts": attempt} time.sleep(2 * attempt) return {"status": "failed", "error": str(last_exc)[:300] if last_exc else "unknown", "attempts": max_attempts} def _append_jsonl_if(path: Optional[Path], payload: dict) -> None: if path is None: return with path.open("a", encoding="utf-8") as f: f.write(json.dumps(payload, ensure_ascii=False)) f.write("\n") # --------------------------------------------------------------------------- # Comparison (base plate run vs source bible) # --------------------------------------------------------------------------- def find_latest_base_run(parent: Path) -> Optional[Path]: if not parent.exists(): return None candidates = sorted( (p for p in parent.iterdir() if p.is_dir() and p.name.startswith("2026")), reverse=True, ) return candidates[0] if candidates else None @dataclass class ComparisonRow: plate_id: str issue_type: str # luxury_cue_candidate | source_not_supports | source_requires_missing | unknown evidence_quote: str base_evidence: str severity_candidate: str notes: str = "" def build_comparison_report( base_run_dir: Path, bundle: SourceBundle, evidence_rows: list[EvidenceRow], gemini_bible: Optional[dict] = None, ) -> list[ComparisonRow]: rows: list[ComparisonRow] = [] base_prompts_dir = base_run_dir / "prompts" / "base" realized_dir = base_run_dir / "realized_spatial_cards" # source 가 "주장하는" 가구/sub_space (deterministic evidence 기반). source_keywords = {r.keyword for r in evidence_rows} # Gemini bible 의 required + forbidden (있을 때만). bible_required: list[str] = [] bible_forbidden: list[str] = [] if gemini_bible: bible_required = list(gemini_bible.get("required_visual_cues") or []) bible_forbidden = list(gemini_bible.get("forbidden_luxury_cues") or []) # 각 base plate prompt + LVM card 검사. if not base_prompts_dir.exists(): return rows for prompt_path in sorted(base_prompts_dir.glob("*.txt")): stem = prompt_path.stem plate_id = f"base__{stem}" prompt_text = prompt_path.read_text(encoding="utf-8") prompt_lower = prompt_text.lower() realized_path = realized_dir / f"{plate_id}.json" realized: dict = {} if realized_path.exists(): try: realized = json.loads(realized_path.read_text(encoding="utf-8")) except json.JSONDecodeError: realized = {} fixed_labels = [f.get("label", "") for f in realized.get("fixed_objects", [])] # 1. luxury_cue_candidate for luxury in LUXURY_CUE_CANDIDATES + bible_forbidden: l_low = luxury.lower() if l_low in prompt_lower or any(l_low in f.lower() for f in fixed_labels): src_evidence = "Gemini bible.forbidden" if luxury in bible_forbidden else "deterministic list" rows.append(ComparisonRow( plate_id=plate_id, issue_type="luxury_cue_candidate", evidence_quote=luxury, base_evidence=f"prompt or LVM contains {luxury!r}", severity_candidate="suspicious", notes=src_evidence, )) # 2. source_not_supports — LVM fixed_object 가 source 에 evidence 0 for label in fixed_labels: # 매우 일반 명사는 스킵 if not label or len(label) < 2: continue label_norm = label.lower().replace("_", " ").strip() # 한국어 source evidence keyword 와 매칭 시도 (단순 — 의미상 1:1 매칭 N/A) # 빈약한 evidence: deterministic keyword 의 어떤 한국어와도 substring 매칭 없으면. has_source_evidence = False for kw in source_keywords: if kw in label_norm or label_norm in kw: has_source_evidence = True break # 영문 → 일반화 키워드 매핑 약하므로 candidate 만 표시. if not has_source_evidence: rows.append(ComparisonRow( plate_id=plate_id, issue_type="source_not_supports", evidence_quote=label, base_evidence=f"LVM fixed_object label='{label}'", severity_candidate="info", notes="deterministic match 0 — 의미 등가 수동 확인 필요", )) # 3. source_requires_missing — Gemini bible.required_visual_cues 중 LVM 에 없음 for required in bible_required: r_low = required.lower() if not any(r_low in f.lower() for f in fixed_labels): rows.append(ComparisonRow( plate_id=plate_id, issue_type="source_requires_missing", evidence_quote=required, base_evidence=f"LVM fixed_objects 에 '{required}' 없음", severity_candidate="likely_mismatch", notes="Gemini bible.required_visual_cues 미반영", )) return rows # --------------------------------------------------------------------------- # Webserver (Phase D 와 동일 패턴) # --------------------------------------------------------------------------- def _find_free_port(start: int, attempts: int = 6, bind: str = "127.0.0.1") -> Optional[int]: import socket as _socket # noqa: PLC0415 for offset in range(attempts): port = start + offset s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) try: s.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1) s.bind((bind, port)) s.close() return port except OSError: continue return None def start_static_webserver(run_dir: Path, port: int, bind: str = "127.0.0.1") -> dict: import subprocess # noqa: PLC0415 free = _find_free_port(port, attempts=6, bind=bind) if free is None: return {"status": "no_free_port", "tried_from": port} log_path = run_dir / "_serve.log" log_fh = log_path.open("w", encoding="utf-8") proc = subprocess.Popen( # noqa: S603 [sys.executable, "-m", "http.server", str(free), "--bind", bind], cwd=str(run_dir), stdout=log_fh, stderr=log_fh, ) time.sleep(0.6) rc = proc.poll() if rc is not None: try: log_tail = log_path.read_text(encoding="utf-8", errors="replace")[-400:] except Exception: log_tail = "" return {"status": "died_early", "pid": proc.pid, "port": free, "bind": bind, "exit_code": rc, "log": str(log_path), "log_tail": log_tail} return {"status": "started", "pid": proc.pid, "port": free, "bind": bind, "log": str(log_path), "cwd": str(run_dir)} # --------------------------------------------------------------------------- # Writers # --------------------------------------------------------------------------- def _safe_tsv(v) -> str: return str(v).replace("\t", " ").replace("\r", " ").replace("\n", "\\n") def write_tsv(path: Path, rows: list[list]) -> None: with path.open("w", encoding="utf-8") as f: for r in rows: f.write("\t".join(_safe_tsv(c) for c in r)) f.write("\n") def _short(s: str, limit: int = HTML_QUOTE_LIMIT) -> str: return s if len(s) <= limit else s[:limit] + "..." def render_html( run_dir: Path, bundle: SourceBundle, evidence_rows: list[EvidenceRow], base_run_dir: Optional[Path], comparison_rows: list[ComparisonRow], gemini_bible: Optional[dict], run_meta: dict, ) -> None: # §1 source bundle missing_rows = "".join( f"{html.escape(m['key'])}" f"{html.escape(m['looked_at'])}" f"{html.escape(m['reason'])}" for m in bundle.missing_inputs ) # §2 evidence — category 별 grouping by_cat: dict[str, list[EvidenceRow]] = {} for r in evidence_rows: by_cat.setdefault(r.category, []).append(r) evidence_html_parts = [] for cat in ("sub_space", "fixture", "state", "condition_socio"): items = by_cat.get(cat, []) if not items: continue rows_html = "".join( "" f"{html.escape(r.source)}" f"{html.escape(r.keyword)}" f"{html.escape(r.candidate_field)}" f"{html.escape(str(r.char_offset))}" f"{html.escape(_short(r.quote))}" f"{html.escape(r.confidence)}" "" for r in items[:60] ) more = f" (+{len(items)-60} more, see TSV)" if len(items) > 60 else "" evidence_html_parts.append( f"

{html.escape(cat)} ({len(items)}{more})

" "" "" "" "" + rows_html + "
sourcekeywordcandidate_fieldoffsetquoteconfidence
" ) evidence_html = "".join(evidence_html_parts) or "

(매칭 0)

" # §3 Gemini bible if gemini_bible: gemini_html = ( f"
raw JSON ({len(json.dumps(gemini_bible))} chars)" f"
{html.escape(json.dumps(gemini_bible, ensure_ascii=False, indent=2))}
" "
" ) else: gemini_html = "

미실행 — `--generate` 명시 필요.

" # §4 base plate inline — Codex Blocking 1: run_dir 안 base_plate_assets// 사용. base_section = "" asset_rel = run_meta.get("base_plate_assets_dir") # 상대 path (run_dir 기준) if base_run_dir and base_run_dir.exists() and asset_rel: asset_dir = run_dir / asset_rel plate_files = sorted(asset_dir.glob("*.png")) if asset_dir.exists() else [] cards = [] for png in plate_files: stem = png.stem plate_id = f"base__{stem}" rel = f"{asset_rel}/{png.name}" # 상대 path (webserver root = run_dir) lvm_path = asset_dir / f"{plate_id}.json" lvm_summary = "" if lvm_path.exists(): try: card = json.loads(lvm_path.read_text(encoding="utf-8")) band = card.get("overall_confidence_band", "?") fixed = [f["label"] for f in card.get("fixed_objects", [])[:5]] lvm_summary = f"band={band} | fixed={', '.join(fixed)}" except json.JSONDecodeError: lvm_summary = "(LVM JSON parse error)" cards.append( "
" f'' f'{html.escape(plate_id)}' f'
' f'
{html.escape(plate_id)}
' f'
LVM: {html.escape(lvm_summary)}
' "
" ) base_section = ( f"

copied from {html.escape(str(base_run_dir))} " f"→ {html.escape(asset_rel)}/

" + "".join(cards) ) if cards else "

(base plate 없음)

" else: base_section = ( f"

base run dir 없음 (찾은 위치: " f"{html.escape(str(base_run_dir))}).

" if base_run_dir else "

base run dir 미지정.

" ) # §5 comparison — Codex Q5: plate_id grouping plate_groups: dict[str, list[ComparisonRow]] = {} for c in comparison_rows: plate_groups.setdefault(c.plate_id, []).append(c) comp_blocks = [] for plate_id in sorted(plate_groups): items = plate_groups[plate_id] by_issue: dict[str, list[ComparisonRow]] = {} for r in items: by_issue.setdefault(r.issue_type, []).append(r) header = ( f"{html.escape(plate_id)} — total {len(items)} " + ", ".join(f"{k}: {len(v)}" for k, v in by_issue.items()) + "" ) body_rows = "".join( "" f"{html.escape(r.issue_type)}" f"{html.escape(_short(r.evidence_quote, 80))}" f"{html.escape(_short(r.base_evidence, 100))}" f"{html.escape(r.severity_candidate)}" f"{html.escape(_short(r.notes, 80))}" "" for r in items ) comp_blocks.append( f"
{header}" "" "" f"{body_rows}
issueevidence_quotebase_evidenceseveritynotes
" ) comp_html = "".join(comp_blocks) or "

(comparison 없음 — base run 없거나 Gemini 미실행)

" serve_info = run_meta.get("serve_info") or {} serve_html = "" if serve_info.get("status") == "started": url = f"http://{serve_info['bind']}:{serve_info['port']}/index.html" serve_html = ( f"

webserver: {html.escape(url)} " f"(PID {serve_info['pid']}, log {html.escape(serve_info['log'])})

" ) body = f""" Rooftop Source Grounding — {html.escape(run_meta['run_id'])}

Rooftop Source Grounding Experiment

run_id {html.escape(run_meta['run_id'])} · plan v1

project {html.escape(PROJECT_ID)} · episode {html.escape(EPISODE_ID)} · base_run {html.escape(str(base_run_dir) if base_run_dir else '(none)')}

{serve_html}

1. Source bundle

episode fulltext: {bundle.episode_meta.get('fulltext_length', 0)} chars · planning_doc: {bundle.project_meta.get('planning_doc_length', 0)} chars · missing inputs: {len(bundle.missing_inputs)}

{('' + missing_rows + '
keylooked_atreason
') if missing_rows else '

(missing 없음)

'}

2. Deterministic source evidence ({len(evidence_rows)})

{evidence_html}

3. Gemini rooftop bible

{gemini_html}

4. Current base plate (Phase D run, read-only)

{base_section}

5. Comparison candidates ({len(comparison_rows)})

판정 X — evidence mismatch 후보만 표시. 사람이 검토 후 결정.

{comp_html}

6. Run meta

{html.escape(json.dumps({k: v for k, v in run_meta.items() if k != 'serve_info'}, ensure_ascii=False, indent=2))}
""" (run_dir / "index.html").write_text(body, encoding="utf-8") def copy_base_plate_assets( run_dir: Path, base_run_dir: Optional[Path] ) -> Optional[Path]: """Codex Blocking 1: base PNG + LVM JSON 을 run_dir 안에 copy → webserver inline 가능. Returns 상대 path prefix (run_dir 기준) — HTML img src 에 사용. """ import shutil # noqa: PLC0415 if base_run_dir is None or not base_run_dir.exists(): return None dest = run_dir / "base_plate_assets" / base_run_dir.name dest.mkdir(parents=True, exist_ok=True) plates_src = base_run_dir / "base_plates" if plates_src.exists(): for png in plates_src.glob("*.png"): shutil.copy2(png, dest / png.name) cards_src = base_run_dir / "realized_spatial_cards" if cards_src.exists(): for j in cards_src.glob("*.json"): shutil.copy2(j, dest / j.name) return dest def write_outputs( run_dir: Path, bundle: SourceBundle, evidence_rows: list[EvidenceRow], base_run_dir: Optional[Path], comparison_rows: list[ComparisonRow], gemini_bible: Optional[dict], run_meta: dict, ) -> None: run_dir.mkdir(parents=True, exist_ok=True) # Copy base plate assets BEFORE rendering HTML (Codex Blocking 1). asset_dir = copy_base_plate_assets(run_dir, base_run_dir) run_meta["base_plate_assets_dir"] = str(asset_dir.relative_to(run_dir)) if asset_dir else None # source_bundle.md (full text — quote limit X) md_parts = [ f"# Source bundle — {run_meta['run_id']}", "", f"project: `{PROJECT_ID}`", f"episode: `{EPISODE_ID}`", f"episode meta: `{json.dumps(bundle.episode_meta, ensure_ascii=False)}`", f"project meta: `{json.dumps(bundle.project_meta, ensure_ascii=False)}`", f"missing inputs: `{json.dumps(bundle.missing_inputs, ensure_ascii=False)}`", "", "## EPISODE 1 FULLTEXT", "```", bundle.episode_fulltext or "(empty)", "```", "", "## PROJECT PLANNING DOC", "```", bundle.planning_doc_text or "(empty)", "```", ] (run_dir / "source_bundle.md").write_text("\n".join(md_parts), encoding="utf-8") (run_dir / "source_bundle.json").write_text(json.dumps({ "episode_meta": bundle.episode_meta, "project_meta": bundle.project_meta, "missing_inputs": bundle.missing_inputs, "episode_fulltext": bundle.episode_fulltext, "planning_doc_text": bundle.planning_doc_text, }, ensure_ascii=False, indent=2), encoding="utf-8") # source_evidence.tsv + json header = ["source", "keyword", "category", "candidate_field", "char_offset", "confidence", "quote", "notes"] rows: list[list] = [header] for r in evidence_rows: rows.append([r.source, r.keyword, r.category, r.candidate_field, r.char_offset, r.confidence, r.quote, r.notes]) write_tsv(run_dir / "source_evidence.tsv", rows) (run_dir / "source_evidence.json").write_text(json.dumps( [r.__dict__ for r in evidence_rows], ensure_ascii=False, indent=2 ), encoding="utf-8") # comparison_report.tsv + json if comparison_rows: c_header = ["plate_id", "issue_type", "severity_candidate", "evidence_quote", "base_evidence", "notes"] c_rows: list[list] = [c_header] for c in comparison_rows: c_rows.append([c.plate_id, c.issue_type, c.severity_candidate, c.evidence_quote, c.base_evidence, c.notes]) write_tsv(run_dir / "comparison_report.tsv", c_rows) (run_dir / "comparison_report.json").write_text(json.dumps( [c.__dict__ for c in comparison_rows], ensure_ascii=False, indent=2 ), encoding="utf-8") else: (run_dir / "comparison_report.tsv").write_text("(empty)\n", encoding="utf-8") # gemini_rooftop_bible.json if gemini_bible: (run_dir / "gemini_rooftop_bible.json").write_text( json.dumps(gemini_bible, ensure_ascii=False, indent=2), encoding="utf-8" ) (run_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8" ) render_html(run_dir, bundle, evidence_rows, base_run_dir, comparison_rows, gemini_bible, run_meta) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def parse_args() -> argparse.Namespace: ap = argparse.ArgumentParser(description="Rooftop Source Grounding Experiment") ap.add_argument("--run-id", default=None) ap.add_argument("--generate", action="store_true", help="Gemini bible 실호출 허용 (default OFF — dry-run)") ap.add_argument("--gemini-model", default=GEMINI_FLASH_DEFAULT) ap.add_argument("--base-run-dir", default=None, help="비교 대상 base plate run dir. default = 가장 최근") ap.add_argument("--output-base", default=str(DEFAULT_OUTPUT_DIR)) ap.add_argument("--no-serve", action="store_true") ap.add_argument("--serve-port", type=int, default=8766, help="webserver 포트 (Phase D 8765 와 충돌 회피)") ap.add_argument("--bind", default="127.0.0.1") ap.add_argument("--estimate-only", action="store_true") return ap.parse_args() def main() -> int: args = parse_args() if args.bind == "0.0.0.0": print("WARNING: --bind 0.0.0.0 — unauthenticated local network server.", file=sys.stderr) run_id = args.run_id or f"{datetime.now():%Y%m%d_%H%M}_{uuid.uuid4().hex[:6]}" output_base = Path(args.output_base).resolve() run_dir = output_base / run_id with SessionLocal() as session: bundle = load_source_bundle(session) evidence_rows = extract_evidence(bundle) # base run dir 결정 if args.base_run_dir: base_run_dir: Optional[Path] = Path(args.base_run_dir).resolve() if not base_run_dir.exists(): print(f"WARNING: --base-run-dir {base_run_dir} 부재. comparison 빈 채로 진행.", file=sys.stderr) base_run_dir = None else: base_run_dir = find_latest_base_run(DEFAULT_BASE_RUN_PARENT.resolve()) cost_log = run_dir / "gemini_cost_log.jsonl" call_meta_log = run_dir / "gemini_call_meta.jsonl" # Gemini call (--generate 시만) gemini_bible: Optional[dict] = None gemini_status = "skipped (dry-run)" if args.generate: if not bundle.episode_fulltext or not bundle.planning_doc_text: print("ERROR: --generate 있는데 episode_fulltext / planning_doc_text 비어 있음.", file=sys.stderr) return 4 run_dir.mkdir(parents=True, exist_ok=True) print(f"[gemini] calling {args.gemini_model} (episode={len(bundle.episode_fulltext)} + " f"planning={len(bundle.planning_doc_text)} chars)...", flush=True) result = call_gemini_bible( episode_text=bundle.episode_fulltext, planning_text=bundle.planning_doc_text, gemini_model=args.gemini_model, cost_log_path=cost_log, call_meta_log_path=call_meta_log, ) gemini_status = result.get("status", "failed") if result.get("status") == "ok": gemini_bible = result.get("bible") print(f"[gemini] OK ({result.get('latency_ms', '?')}ms)") else: print(f"[gemini] FAILED: {result.get('error', '?')[:200]}", file=sys.stderr) comparison_rows = build_comparison_report( base_run_dir, bundle, evidence_rows, gemini_bible, ) if base_run_dir else [] serve_info = None if not args.no_serve: run_dir.mkdir(parents=True, exist_ok=True) serve_info = start_static_webserver(run_dir, args.serve_port, bind=args.bind) run_meta = { "run_id": run_id, "created_at": datetime.now().isoformat(timespec="seconds"), "plan_version": "v1", "project_id": PROJECT_ID, "episode_id": EPISODE_ID, "location_short_id": L05_SHORT_ID, "evidence_count": len(evidence_rows), "comparison_count": len(comparison_rows), "base_run_dir": str(base_run_dir) if base_run_dir else None, "gemini_status": gemini_status, "gemini_model": args.gemini_model, "args": { "generate": args.generate, "gemini_model": args.gemini_model, "base_run_dir": str(base_run_dir) if base_run_dir else None, "no_serve": args.no_serve, "serve_port": args.serve_port, "bind": args.bind, }, "serve_info": serve_info, } if args.estimate_only: print(json.dumps(run_meta, ensure_ascii=False, indent=2)) return 0 write_outputs(run_dir, bundle, evidence_rows, base_run_dir, comparison_rows, gemini_bible, run_meta) print(f"\nOK: run 완료 → {run_dir}") print(f" evidence={len(evidence_rows)}, comparison={len(comparison_rows)}, " f"gemini={gemini_status}, missing_inputs={len(bundle.missing_inputs)}") print(f" index: {run_dir / 'index.html'}") if serve_info and serve_info.get("status") == "started": url = f"http://{serve_info['bind']}:{serve_info['port']}/index.html" print(f" webserver: {url} (PID {serve_info['pid']})") print(f" stop: kill {serve_info['pid']}") elif serve_info and serve_info.get("status") == "died_early": print(f" webserver: DIED_EARLY exit={serve_info.get('exit_code')}", file=sys.stderr) # Codex Important 2: --generate 인데 gemini 실패면 nonzero exit (핵심 산출물 실패). if args.generate and gemini_status != "ok": print(f" gemini_status={gemini_status} → exit 9 (Gemini wave 의 핵심 산출물 실패)", file=sys.stderr) return 9 return 0 if __name__ == "__main__": raise SystemExit(main())