#!/usr/bin/env python3 """Background topology planner experiment — W1 dry-run (2026-05-24). scripts_output/background_topology_planner_experiment// read-only: DB write 0 / network/image 모듈 0 / production code 0 / sibling experiment import 0. Goal (plan.md W0b APPROVED_FOR_W1): topology-first generic background topology planner 의 W1 evidence pack 만 수집. topology / shot intent / unit need / ledger 는 후속 wave 에서 채워짐. W1 산출 = source evidence pack + EvidenceExtractionRuleSet + coverage summary + HTML. ★★★ Scope (2026-05-24 사용자 standing rule): 본 실험은 generic background topology planner 검증이다. L05 (rooftop interior) / 옥탑방 / 수리영 / 민숙 / 안방 등은 sample fixture only — 모든 fixture 식별자는 SAMPLE_FIXTURE_* prefix 와 SampleFixtureSpec / EvidenceLexicon 로만 격리. generic collector body 에 sample literal 0. 사용자 standing rule [[feedback-no-scenario-specific-coding]] 참조. Import scope: - app.core.database.SessionLocal (read-only) - app.models.project: Episode, EntityCanon, SceneStill (read-only) - app.models.catalog: ProjectRegistry (read-only) - 표준 라이브러리만 (re, json, csv, html, dataclasses, datetime, pathlib, uuid, argparse, hashlib, sys, os). - 금지: openai / google.genai / google.generativeai / fal_client / fal / PIL / requests / httpx / experiment_* sibling modules. Output 7 files (run dir): input_manifest.json / source_evidence_pack.json / source_evidence.tsv / evidence_extraction_ruleset.json / source_coverage_summary.json / index.html / run_meta.json. """ from __future__ import annotations import argparse import csv import hashlib import html as html_lib import json import os import re import sys import unicodedata import uuid from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Iterable, Optional # repo root + env -------------------------------------------------------------- _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: """Load backend/.env into os.environ so that SessionLocal can connect to PostgreSQL when this script is invoked directly. No overwrite of pre-set env vars. Read-only. """ 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() # ============================================================================ # Plan metadata # ============================================================================ PLAN_VERSION = "btp_w1" # ============================================================================ # SAMPLE FIXTURE (L05 rooftop interior) — generic methodology 검증용 표본 only. # Generic rule body / function signature 에 직접 박지 말 것. 모든 sample 값은 # 이 블록 + SampleFixtureSpec / EvidenceLexicon 안에만 존재. # ============================================================================ SAMPLE_FIXTURE_PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19" SAMPLE_FIXTURE_EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c" SAMPLE_FIXTURE_L05_CANON_ID = "3afbc7a8-b919-4431-a691-0a99057a26ca" SAMPLE_FIXTURE_L05_SHORT_ID = "L05" SAMPLE_FIXTURE_SOURCE_RUN = Path( "scripts_output/rooftop_source_grounding/codex_entry_sanity_gemini_ok" ) SAMPLE_FIXTURE_SOURCE_BIBLE_FILENAME = "gemini_rooftop_bible.json" # Legacy aliases — kept for stability in case tests reference shorter names. SAMPLE_FIXTURE_SHORT_ID = SAMPLE_FIXTURE_L05_SHORT_ID SAMPLE_FIXTURE_CANON_ID = SAMPLE_FIXTURE_L05_CANON_ID DEFAULT_OUTPUT_DIR = Path("scripts_output/background_topology_planner_experiment") DEFAULT_SOURCE_RUN = SAMPLE_FIXTURE_SOURCE_RUN # ---------------------------------------------------------------------------- # Sample fixture lexicon (data-only, sample-specific). Generic engine reads # this exclusively via SampleFixtureSpec.evidence_lexicon — never directly. # Korean terms appearing here (옥상 / 거실 / 안방 / 침실 / 수리영의 방 / 민숙의 # 방 / 욕실 / 주방 / 현관) are fixture vocabulary. Other fixtures will define # their own EvidenceLexicon with different terms. # ---------------------------------------------------------------------------- SAMPLE_FIXTURE_L05_PLACE_TERMS: list[tuple[str, str]] = [ ("옥상", "place_hint"), ("rooftop", "place_hint"), ("옥탑", "place_hint"), ] SAMPLE_FIXTURE_L05_SPACE_TERMS: list[tuple[str, str]] = [ ("거실", "space_hint"), ("주방", "space_hint"), ("싱크대", "furniture_hint"), ("식탁", "furniture_hint"), ("침실", "space_hint"), ("안방", "space_hint"), ("수리영의 방", "space_hint"), ("수리영의방", "space_hint"), ("민숙의 방", "space_hint"), ("욕실", "space_hint"), ("화장실", "space_hint"), ("현관", "boundary_hint"), ("문", "door_window_hint"), ("창문", "door_window_hint"), ("미닫이창", "door_window_hint"), ("커튼", "furniture_hint"), ("천장", "zone_hint"), ("벽면", "zone_hint"), ("바닥", "zone_hint"), ("계단", "zone_hint"), ] SAMPLE_FIXTURE_L05_STATE_TERMS: list[tuple[str, str]] = [ ("황혼", "state_hint"), ("해질", "state_hint"), ("새벽", "state_hint"), ("동틀", "state_hint"), ("어두운", "state_hint"), ("어둠", "state_hint"), ("깨끗", "state_hint"), ("정돈", "state_hint"), ("어지럽", "state_hint"), ("뒤집힌", "state_hint"), ("핏자국", "state_hint"), ("시신", "state_hint"), ("표식", "state_hint"), ] SAMPLE_FIXTURE_L05_CAMERA_TERMS: list[tuple[str, str]] = [ ("wide", "camera_hint"), ("광각", "camera_hint"), ("close-up", "camera_hint"), ("클로즈업", "camera_hint"), ("doorway", "camera_hint"), ("문턱", "camera_hint"), ("macro", "camera_hint"), ("매크로", "camera_hint"), ("eye-level", "camera_hint"), ("눈높이", "camera_hint"), ("pan", "movement_hint"), ("track", "movement_hint"), ("dolly", "movement_hint"), ] # Generic builtin lexicon — sample-agnostic 도메인 일반어. Any new fixture # inherits these terms automatically via build_evidence_extraction_ruleset. GENERIC_BUILTIN_PLACE_TERMS: list[tuple[str, str]] = [ ("building", "place_hint"), ("interior", "set_hint"), ("exterior", "set_hint"), ] GENERIC_BUILTIN_SPACE_TERMS: list[tuple[str, str]] = [ ("room", "space_hint"), ("window", "door_window_hint"), ("door", "door_window_hint"), ("balcony", "boundary_hint"), ("stairs", "zone_hint"), ] GENERIC_BUILTIN_STATE_TERMS: list[tuple[str, str]] = [ ("night", "state_hint"), ("morning", "state_hint"), ("dusk", "state_hint"), ] GENERIC_BUILTIN_CAMERA_TERMS: list[tuple[str, str]] = [ ("medium shot", "camera_hint"), ("establishing", "camera_hint"), ("tracking shot", "movement_hint"), ] # ============================================================================ # W1c review-cockpit constants (plan §8-E) — generic, sample-agnostic. # Korean negation phrases are language-level (not L05-specific). bare `안 ` # 은 `안방` 오탐 위험이라 금지 (Codex W1c review 명시). # ============================================================================ GENERIC_NEGATION_TERMS_KOREAN: list[str] = [ "없이", "없", "않", "아니", "금지", "제외", "배제", "보이지", "나타나지", "흔적을 찾지", "감쪽같이", "감쪽 같이", "사라졌", "사라진", ] # English negation = regex with word boundaries to avoid `know` ⊃ `no` 오탐. GENERIC_NEGATION_REGEX_ENGLISH: list[str] = [ r"\bno\b", r"\bnot\b", r"\bwithout\b", r"\black\b", r"\blacks\b", r"\babsent\b", r"\babsence\b", r"\bexcluded\b", r"\bexclude\b", ] # Compiled lazily into _ENGLISH_NEGATION_PATTERN at module level. _ENGLISH_NEGATION_PATTERN = re.compile( "|".join(GENERIC_NEGATION_REGEX_ENGLISH), flags=re.IGNORECASE, ) # Default readiness threshold dictionary (per-dimension). Each value = # {: }. SampleFixtureSpec ships with this as default; # new fixtures may override by passing their own readiness_thresholds. DEFAULT_READINESS_THRESHOLDS: dict[str, dict[str, Any]] = { "topology_candidate": { # place / space are *distinct match-term* counts. "place_distinct_terms": 1, "space_distinct_terms": 4, # boundary / door-window are *row* counts (the underlying lexicon # rarely yields >1 distinct term, so row count is the actually # meaningful coverage signal — plan §8-F W1d IMPORTANT 2). "boundary_rows": 1, "door_window_rows": 1, }, "state_model": { "state_distinct_terms": 3, "max_noisy_ratio": 0.5, }, "camera_intent": { "camera_distinct_terms": 2, "movement_rows": 1, }, } # ============================================================================ # Data classes # ============================================================================ @dataclass class EvidenceLexicon: """Sample-specific lexicon terms grouped by purpose. Generic builder augments these with `GENERIC_BUILTIN_*` term tables. Each entry is a (term, evidence_type) pair. """ place_terms: list[tuple[str, str]] = field(default_factory=list) space_terms: list[tuple[str, str]] = field(default_factory=list) state_terms: list[tuple[str, str]] = field(default_factory=list) camera_terms: list[tuple[str, str]] = field(default_factory=list) # 확장 슬롯 — future fixture 가 새 evidence_type 추가 가능. extra_terms: list[tuple[str, str]] = field(default_factory=list) @dataclass class SampleFixtureSpec: """All sample-specific data for a single fixture (e.g. L05 rooftop). Generic engine functions (`load_*`, `collect_*`, `build_evidence_*`) MUST take `spec` and read fixture identifiers only from it — never reference SAMPLE_FIXTURE_* module-level constants directly. W1c (plan §8-E) adds two review-cockpit fields: - `readiness_thresholds`: dimension → metric thresholds dict - `negation_terms`: noisy-heuristic vocabulary (Korean phrase + opt. English regex literal); generic defaults applied per fixture. """ fixture_id: str project_id: str episode_id: str canon_id: str location_short_id: str source_run_path: Path source_bible_filename: str evidence_lexicon: EvidenceLexicon readiness_thresholds: dict[str, dict[str, Any]] = field( default_factory=lambda: { dim: dict(metrics) for dim, metrics in DEFAULT_READINESS_THRESHOLDS.items() }, ) # Negation lexicon — defaults to the generic Korean + English union; a # fixture may pass an alternative list (e.g. domain dialect terms). negation_terms: list[str] = field( default_factory=lambda: list(GENERIC_NEGATION_TERMS_KOREAN) + list(GENERIC_NEGATION_REGEX_ENGLISH), ) @dataclass class SourcePack: """In-memory bundle of all read-only DB / artifact inputs. The collector layer consumes this; main() builds it via load_* DB readers. """ planning_doc_text: str = "" episode_fulltext: str = "" selected_shots: list[dict] = field(default_factory=list) entity_catalog: list[dict] = field(default_factory=list) location_catalog: list[dict] = field(default_factory=list) # existing_artifacts: list of (relative_path, parsed_object). existing_artifacts: list[tuple[str, Any]] = field(default_factory=list) # diagnostic_artifacts: same shape — only present when # --include-diagnostic-runs is set; collector emits them as # source_kind="existing_artifact" rows that augment but never modify the # base evidence count of other source kinds (IMPORTANT 4). diagnostic_artifacts: list[tuple[str, Any]] = field(default_factory=list) missing_inputs: list[dict] = field(default_factory=list) # ============================================================================ # Sample-fixture loader (L05 rooftop interior) — SAMPLE-SPECIFIC. 새 location # 추가 시 `build_sample_fixture__spec()` 새로 정의하면 됨. # ============================================================================ def build_sample_fixture_l05_spec() -> SampleFixtureSpec: """Construct the L05 rooftop interior sample fixture spec.""" return SampleFixtureSpec( fixture_id="l05_rooftop_interior", project_id=SAMPLE_FIXTURE_PROJECT_ID, episode_id=SAMPLE_FIXTURE_EPISODE_ID, canon_id=SAMPLE_FIXTURE_L05_CANON_ID, location_short_id=SAMPLE_FIXTURE_L05_SHORT_ID, source_run_path=SAMPLE_FIXTURE_SOURCE_RUN, source_bible_filename=SAMPLE_FIXTURE_SOURCE_BIBLE_FILENAME, evidence_lexicon=EvidenceLexicon( place_terms=list(SAMPLE_FIXTURE_L05_PLACE_TERMS), space_terms=list(SAMPLE_FIXTURE_L05_SPACE_TERMS), state_terms=list(SAMPLE_FIXTURE_L05_STATE_TERMS), camera_terms=list(SAMPLE_FIXTURE_L05_CAMERA_TERMS), ), ) # ============================================================================ # Helpers — hashing, evidence_id, sentence-window quote, normalization # ============================================================================ def _sha256_hex(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def _normalize_quote(quote: str) -> str: """Stable normalization for evidence_id hashing: NFC + whitespace collapse + strip. NB: deterministic across runs — never depends on hash seed. """ if not quote: return "" n = unicodedata.normalize("NFC", quote) # collapse whitespace runs (spaces, tabs, newlines) to single space. n = re.sub(r"\s+", " ", n).strip() return n def _canonical_source_span_json(source_span: dict) -> str: """Return deterministic JSON for the source_span hash anchor. Strips the `source_hash` field itself (which is body-content driven) and any None values so that adding optional context fields later doesn't drift the canonical form. """ norm: dict[str, Any] = {} for k, v in source_span.items(): if k == "source_hash": continue if v is None: continue norm[k] = v return json.dumps(norm, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _make_evidence_id(*, source_kind: str, source_ref: str, source_span: dict, evidence_type: str, normalized_quote: str) -> str: """Deterministic evidence_id = "ev_" + sha256(anchor)[:16]. The anchor mixes source_kind, source_ref, the canonical source_span JSON (without source_hash, see _canonical_source_span_json), evidence_type, and normalized_quote — see plan §2-C BLOCKING 1. """ anchor = "|".join([ source_kind, source_ref, _canonical_source_span_json(source_span), evidence_type, normalized_quote, ]) return "ev_" + hashlib.sha256(anchor.encode("utf-8")).hexdigest()[:16] _SENTENCE_BOUNDARY = re.compile(r"[.!?\n]") def _extract_sentence_window(text: str, idx: int, k_len: int, max_chars: int = 280) -> str: """Return a ≤ max_chars sentence-bounded window around idx..idx+k_len. Boundaries are the nearest `.`, `!`, `?`, or newline. Falls back to a half-window slice if no boundary is found. """ if not text: return "" start = max(0, idx) end = min(len(text), idx + max(k_len, 1)) # backward search back = -1 for ch in [".", "!", "?", "\n"]: pos = text.rfind(ch, 0, start) if pos > back: back = pos sentence_start = (back + 1) if back >= 0 else max(0, start - max_chars // 2) # forward search forward_candidates: list[int] = [] for ch in [".", "!", "?", "\n"]: pos = text.find(ch, end) if pos >= 0: forward_candidates.append(pos) sentence_end = (min(forward_candidates) + 1) if forward_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 _iter_term_positions(text: str, term: str) -> Iterable[int]: if not term or not text: return [] out: list[int] = [] start = 0 while True: i = text.find(term, start) if i < 0: break out.append(i) start = i + max(1, len(term)) return out # ============================================================================ # EvidenceExtractionRuleSet builder (plan §1-G) # ============================================================================ def _ruleset_rule(*, rule_id: str, evidence_type: str, match_terms: list[str], candidate_contract_targets: list[str], confidence_band_default: str, applies_to_source_kinds: list[str], match_kind: str = "literal_term", match_pattern: Optional[str] = None, negation_terms: Optional[list[str]] = None) -> dict: return { "rule_id": rule_id, "evidence_type": evidence_type, "match_kind": match_kind, "match_terms": list(match_terms), "match_pattern": match_pattern, "candidate_contract_targets": list(candidate_contract_targets), "confidence_band_default": confidence_band_default, "negation_terms": list(negation_terms or []), "applies_to_source_kinds": list(applies_to_source_kinds), } # Contract-target map (evidence_type → candidate fields). Sample-agnostic. _CONTRACT_TARGETS_BY_TYPE: dict[str, list[str]] = { "place_hint": [ "PlaceContinuityGroup.member_location_short_ids", "PlaceContinuityGroup.physical_place_key", ], "set_hint": [ "SetTopologyGraph.continuity_policy", ], "space_hint": [ "SetTopologyGraph.nodes", "ShotSpatialIntent.primary_space_candidates", ], "zone_hint": [ "SetTopologyGraph.nodes", ], "boundary_hint": [ "SetTopologyGraph.nodes", ], "door_window_hint": [ "SetTopologyGraph.edges", ], "furniture_hint": [ "ShotSpatialIntent.visible_secondary_candidates", ], "state_hint": [ "StructuralStateModel.state_layers", ], "camera_hint": [ "ShotSpatialIntent.camera_intent", ], "movement_hint": [ "ShotSpatialIntent.camera_intent", ], "entity_layout_hint": [ "ShotSpatialIntent.visible_secondary_candidates", ], "ambiguity_hint": [ "ShotSpatialIntent.ambiguity_flags", ], } def _compute_lexicon_hash(lexicon: EvidenceLexicon) -> str: """Stable hash of the full lexicon term set (sample + builtin merged). Order-insensitive: terms are sorted before hashing. """ all_terms = sorted({ f"{cat}:{term}:{etype}" for cat, lst in [ ("place", lexicon.place_terms), ("space", lexicon.space_terms), ("state", lexicon.state_terms), ("camera", lexicon.camera_terms), ("extra", lexicon.extra_terms), ("builtin_place", GENERIC_BUILTIN_PLACE_TERMS), ("builtin_space", GENERIC_BUILTIN_SPACE_TERMS), ("builtin_state", GENERIC_BUILTIN_STATE_TERMS), ("builtin_camera", GENERIC_BUILTIN_CAMERA_TERMS), ] for term, etype in lst }) return _sha256_hex("\n".join(all_terms)) def build_evidence_extraction_ruleset(spec: SampleFixtureSpec) -> dict: """Merge sample fixture EvidenceLexicon with generic builtin term tables and produce the §1-G EvidenceExtractionRuleSet JSON dict. The ruleset is sample-agnostic in shape — only `match_terms` content differs per fixture. Every collector consumes this exclusively, never looking at SAMPLE_FIXTURE_* constants directly. """ text_source_kinds = [ "planning_doc", "episode_fulltext", "shot_description", "scene_summary", ] catalog_source_kinds = ["entity_catalog", "location_catalog"] artifact_source_kinds = ["existing_artifact"] all_text_kinds = text_source_kinds + catalog_source_kinds + artifact_source_kinds rules: list[dict] = [] # 1) Per-fixture lexicon entries (with builtin merged in). grouped: list[tuple[str, list[tuple[str, str]]]] = [ ("fixture_place", list(spec.evidence_lexicon.place_terms)), ("fixture_space", list(spec.evidence_lexicon.space_terms)), ("fixture_state", list(spec.evidence_lexicon.state_terms)), ("fixture_camera", list(spec.evidence_lexicon.camera_terms)), ("fixture_extra", list(spec.evidence_lexicon.extra_terms)), ("generic_builtin_place", list(GENERIC_BUILTIN_PLACE_TERMS)), ("generic_builtin_space", list(GENERIC_BUILTIN_SPACE_TERMS)), ("generic_builtin_state", list(GENERIC_BUILTIN_STATE_TERMS)), ("generic_builtin_camera", list(GENERIC_BUILTIN_CAMERA_TERMS)), ] for rule_prefix, term_list in grouped: if not term_list: continue # Group terms by evidence_type so each rule_id collects a coherent set. by_type: dict[str, list[str]] = {} for term, etype in term_list: by_type.setdefault(etype, []).append(term) for etype, terms in sorted(by_type.items()): confidence = ( "observed" if rule_prefix.startswith("fixture_") and etype in {"space_hint", "door_window_hint", "boundary_hint", "furniture_hint", "zone_hint"} else "inferred_candidate" ) rules.append(_ruleset_rule( rule_id=f"{rule_prefix}_{etype}", evidence_type=etype, match_terms=sorted(set(terms)), candidate_contract_targets=_CONTRACT_TARGETS_BY_TYPE.get( etype, []), confidence_band_default=confidence, applies_to_source_kinds=list(all_text_kinds), )) # 2) Structural rule — visible_entities_json on shot_description rows. rules.append(_ruleset_rule( rule_id="shot_visible_entities", evidence_type="entity_layout_hint", match_terms=[], match_kind="structural_field", match_pattern="$.visible_entities_json", candidate_contract_targets=[ "ShotSpatialIntent.visible_secondary_candidates", "ShotSpatialIntent.loc_short_ids", ], confidence_band_default="observed", applies_to_source_kinds=["shot_description"], )) # 3) Structural rule — entity description / metadata pull. rules.append(_ruleset_rule( rule_id="entity_description_pull", evidence_type="entity_layout_hint", match_terms=[], match_kind="row_field_value", match_pattern="description", candidate_contract_targets=[ "PlaceContinuityGroup.member_location_short_ids", "ShotSpatialIntent.visible_secondary_candidates", ], confidence_band_default="observed", applies_to_source_kinds=["entity_catalog", "location_catalog"], )) # 4) Structural rule — existing artifact JSON pointer pull. rules.append(_ruleset_rule( rule_id="existing_artifact_pull", evidence_type="space_hint", match_terms=[], match_kind="json_pointer", match_pattern="$.sub_spaces[*].name", candidate_contract_targets=[ "SetTopologyGraph.nodes", ], confidence_band_default="inferred_candidate", applies_to_source_kinds=["existing_artifact"], )) lex_hash = _compute_lexicon_hash(spec.evidence_lexicon) return { "ruleset_id": f"btp_w1_{spec.fixture_id}", "ruleset_version": "1.0", "lexicon_hash": lex_hash, "rules": rules, } # ============================================================================ # Generic collectors — all take (ruleset, lexicon, spec). No fixture literal # in body. Each collector dedupes the 5-tuple identity (source_kind, # source_ref, source_span canonical, evidence_type, normalized_quote). # ============================================================================ def _split_basis(value: Any) -> list[str]: """Decompose an inference_basis value into distinct non-empty segments. Accepts both raw string ("a; b") and pre-merged form. """ if not value: return [] if isinstance(value, (list, tuple, set)): parts = [str(v).strip() for v in value] else: parts = [seg.strip() for seg in str(value).split(";")] return [p for p in parts if p] def _dedup_merge(rows: list[dict]) -> list[dict]: """Merge rows sharing the same evidence_id by union-merging candidate_contract_targets, extracted_by, match_terms, and inference_basis (distinct sorted, "; " joined — plan §8-D / W1b IMPORTANT 3). """ by_id: dict[str, dict] = {} for row in rows: eid = row["evidence_id"] existing = by_id.get(eid) if existing is None: by_id[eid] = dict(row) existing = by_id[eid] existing["candidate_contract_targets"] = sorted(set( existing.get("candidate_contract_targets") or [], )) extracted = existing.get("extracted_by") extracted_list = ([extracted] if isinstance(extracted, str) else list(extracted or [])) existing["extracted_by"] = sorted(set(extracted_list)) # Normalize basis even on first insert so subsequent joins are # idempotent under "; "-segmented input. basis_parts = _split_basis(existing.get("inference_basis")) existing["inference_basis"] = "; ".join(sorted(set(basis_parts))) continue # merge contract targets. merged_targets = sorted(set( (existing.get("candidate_contract_targets") or []) + (row.get("candidate_contract_targets") or []), )) existing["candidate_contract_targets"] = merged_targets # merge extracted_by. ex_old = existing.get("extracted_by") or [] ex_new = row.get("extracted_by") ex_new_list = ([ex_new] if isinstance(ex_new, str) else list(ex_new or [])) if isinstance(ex_old, str): ex_old = [ex_old] existing["extracted_by"] = sorted(set(list(ex_old) + ex_new_list)) # merge match_terms. merged_terms = sorted(set( (existing.get("match_terms") or []) + (row.get("match_terms") or []), )) existing["match_terms"] = merged_terms # merge inference_basis — distinct sorted, "; " join, empty skipped. merged_basis_parts = ( _split_basis(existing.get("inference_basis")) + _split_basis(row.get("inference_basis")) ) existing["inference_basis"] = "; ".join(sorted(set(merged_basis_parts))) return list(by_id.values()) def _emit_text_row(*, source_kind: str, source_ref: str, text: str, text_hash: str, term: str, evidence_type: str, rule_id: str, confidence_band: str, idx: int, fixture_tags: Optional[list[str]] = None, inference_basis: Optional[str] = None) -> dict: """Build a single EvidenceRow for a text-source match. The caller guarantees that confidence_band is in the enum and that inference_basis is populated when not "observed". """ quote = _extract_sentence_window(text, idx, len(term)) normalized = _normalize_quote(quote) span = { "kind": "text_offset", "char_start": idx, "char_end": idx + len(term), "row_id": None, "row_field": None, "artifact_path": None, "artifact_pointer": None, "source_hash": text_hash, } row = { "evidence_id": _make_evidence_id( source_kind=source_kind, source_ref=source_ref, source_span=span, evidence_type=evidence_type, normalized_quote=normalized, ), "source_kind": source_kind, "source_ref": source_ref, "source_span": span, "quote": quote, "normalized_quote": normalized, "evidence_type": evidence_type, "candidate_contract_targets": list(_CONTRACT_TARGETS_BY_TYPE.get( evidence_type, [])), "confidence_band": confidence_band, "match_terms": [term] if confidence_band != "observed" else [term], "inference_basis": inference_basis or "", "sample_fixture_tags": list(fixture_tags or []), "extracted_by": rule_id, } if confidence_band == "observed": # match_terms optional but we keep it for trace; inference_basis can # be empty string per plan §2-C. row["inference_basis"] = inference_basis or "" return row def _iter_text_rules(ruleset: dict, source_kind: str) -> Iterable[dict]: for rule in ruleset.get("rules", []): if source_kind not in (rule.get("applies_to_source_kinds") or []): continue if rule.get("match_kind") != "literal_term": continue yield rule def _collect_text(*, source_kind: str, source_ref: str, text: str, ruleset: dict, fixture_tags: list[str]) -> list[dict]: if not text: return [] text_hash = _sha256_hex(text) out: list[dict] = [] for rule in _iter_text_rules(ruleset, source_kind): etype = rule["evidence_type"] confidence_band = rule.get("confidence_band_default", "inferred_candidate") rule_id = rule["rule_id"] for term in rule.get("match_terms", []): for idx in _iter_term_positions(text, term): # plan §2-C: inference_basis 는 generic 표현, sample literal # 금지. 실제 term/quote 는 match_terms/quote/source_span 으로 # 트레이스 가능 — 여기서는 rule_id 만 남긴다. inference_basis = ( f"literal lexicon term matched by rule {rule_id}" if confidence_band != "observed" else "" ) out.append(_emit_text_row( source_kind=source_kind, source_ref=source_ref, text=text, text_hash=text_hash, term=term, evidence_type=etype, rule_id=rule_id, confidence_band=confidence_band, idx=idx, fixture_tags=fixture_tags, inference_basis=inference_basis, )) return _dedup_merge(out) def collect_evidence_from_planning_doc(*, text: str, ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: return _collect_text( source_kind="planning_doc", source_ref=f"project:{spec.project_id}", text=text or "", ruleset=ruleset, fixture_tags=[f"fixture:{spec.fixture_id}"], ) def collect_evidence_from_episode_fulltext(*, text: str, ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: return _collect_text( source_kind="episode_fulltext", source_ref=f"episode:{spec.episode_id}", text=text or "", ruleset=ruleset, fixture_tags=[f"fixture:{spec.fixture_id}"], ) def collect_evidence_from_shots(*, shots: list[dict], ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: """Emit text-rule rows over shot_description + scene_summary, plus one structural row per shot for the visible_entities_json contract target. """ out: list[dict] = [] fixture_tags = [f"fixture:{spec.fixture_id}"] for shot in shots or []: still_id = str(shot.get("still_id") or shot.get("id") or "") description = (shot.get("shot_description") or "").strip() summary = (shot.get("scene_summary") or "").strip() out.extend(_collect_text( source_kind="shot_description", source_ref=f"shot:{still_id}", text=description, ruleset=ruleset, fixture_tags=fixture_tags, )) out.extend(_collect_text( source_kind="scene_summary", source_ref=f"shot:{still_id}", text=summary, ruleset=ruleset, fixture_tags=fixture_tags, )) # Structural row — visible_entities_json (row_pointer span). ve_json = shot.get("visible_entities_json") or "[]" ve_hash = _sha256_hex(str(ve_json)) quote = ( ve_json if len(str(ve_json)) <= 280 else str(ve_json)[:277] + "..." ) normalized = _normalize_quote(quote) span = { "kind": "row_pointer", "char_start": None, "char_end": None, "row_id": still_id, "row_field": "visible_entities_json", "artifact_path": None, "artifact_pointer": None, "source_hash": ve_hash, } row = { "evidence_id": _make_evidence_id( source_kind="shot_description", source_ref=f"shot:{still_id}", source_span=span, evidence_type="entity_layout_hint", normalized_quote=normalized, ), "source_kind": "shot_description", "source_ref": f"shot:{still_id}", "source_span": span, "quote": str(quote), "normalized_quote": normalized, "evidence_type": "entity_layout_hint", "candidate_contract_targets": list(_CONTRACT_TARGETS_BY_TYPE.get( "entity_layout_hint", [])), "confidence_band": "observed", "match_terms": [], "inference_basis": "", "sample_fixture_tags": fixture_tags, "extracted_by": "shot_visible_entities", } out.append(row) return _dedup_merge(out) def collect_evidence_from_entity_catalog(*, entities: list[dict], ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: out: list[dict] = [] fixture_tags = [f"fixture:{spec.fixture_id}"] for ent in entities or []: canon_id = str(ent.get("canon_id") or ent.get("id") or "") description = (ent.get("description") or "").strip() if not description: continue desc_hash = _sha256_hex(description) # Run text rules against the description text; reuse the text helper # but rewrite source_ref / source_span to row_pointer. text_rows = _collect_text( source_kind="entity_catalog", source_ref=f"entity:{canon_id}", text=description, ruleset=ruleset, fixture_tags=fixture_tags, ) for r in text_rows: # Convert span to row_pointer (description body). span = { "kind": "row_pointer", "char_start": None, "char_end": None, "row_id": canon_id, "row_field": "description", "artifact_path": None, "artifact_pointer": None, "source_hash": desc_hash, } r["source_span"] = span r["evidence_id"] = _make_evidence_id( source_kind=r["source_kind"], source_ref=r["source_ref"], source_span=span, evidence_type=r["evidence_type"], normalized_quote=r["normalized_quote"], ) out.extend(text_rows) # Always include a row_field_value pull row so the source_kind appears. quote = description if len(description) <= 280 else description[:277] + "..." normalized = _normalize_quote(quote) span = { "kind": "row_pointer", "char_start": None, "char_end": None, "row_id": canon_id, "row_field": "description", "artifact_path": None, "artifact_pointer": None, "source_hash": desc_hash, } out.append({ "evidence_id": _make_evidence_id( source_kind="entity_catalog", source_ref=f"entity:{canon_id}", source_span=span, evidence_type="entity_layout_hint", normalized_quote=normalized, ), "source_kind": "entity_catalog", "source_ref": f"entity:{canon_id}", "source_span": span, "quote": quote, "normalized_quote": normalized, "evidence_type": "entity_layout_hint", "candidate_contract_targets": list(_CONTRACT_TARGETS_BY_TYPE.get( "entity_layout_hint", [])), "confidence_band": "observed", "match_terms": [], "inference_basis": "", "sample_fixture_tags": fixture_tags, "extracted_by": "entity_description_pull", }) return _dedup_merge(out) def collect_evidence_from_location_catalog(*, entities: list[dict], ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: """Mirror of entity_catalog collector but emits source_kind=location_catalog so that IMPORTANT 1 (location_catalog presence) is satisfiable. """ out: list[dict] = [] fixture_tags = [f"fixture:{spec.fixture_id}"] for ent in entities or []: canon_id = str(ent.get("canon_id") or ent.get("id") or "") description = (ent.get("description") or "").strip() # location_catalog source_kind is emitted even with empty description # (description fallback uses name). name = (ent.get("name") or "").strip() body = description or name or "(empty)" body_hash = _sha256_hex(body) text_rows = _collect_text( source_kind="location_catalog", source_ref=f"location:{canon_id}", text=body, ruleset=ruleset, fixture_tags=fixture_tags, ) for r in text_rows: span = { "kind": "row_pointer", "char_start": None, "char_end": None, "row_id": canon_id, "row_field": "description", "artifact_path": None, "artifact_pointer": None, "source_hash": body_hash, } r["source_span"] = span r["evidence_id"] = _make_evidence_id( source_kind=r["source_kind"], source_ref=r["source_ref"], source_span=span, evidence_type=r["evidence_type"], normalized_quote=r["normalized_quote"], ) out.extend(text_rows) quote = body if len(body) <= 280 else body[:277] + "..." normalized = _normalize_quote(quote) span = { "kind": "row_pointer", "char_start": None, "char_end": None, "row_id": canon_id, "row_field": "description", "artifact_path": None, "artifact_pointer": None, "source_hash": body_hash, } out.append({ "evidence_id": _make_evidence_id( source_kind="location_catalog", source_ref=f"location:{canon_id}", source_span=span, evidence_type="entity_layout_hint", normalized_quote=normalized, ), "source_kind": "location_catalog", "source_ref": f"location:{canon_id}", "source_span": span, "quote": quote, "normalized_quote": normalized, "evidence_type": "entity_layout_hint", "candidate_contract_targets": list(_CONTRACT_TARGETS_BY_TYPE.get( "entity_layout_hint", [])), "confidence_band": "observed", "match_terms": [], "inference_basis": "", "sample_fixture_tags": fixture_tags, "extracted_by": "entity_description_pull", }) return _dedup_merge(out) def _walk_strings(obj: Any, pointer: str = "$"): """Yield (pointer_path, value) for each str leaf inside obj.""" if isinstance(obj, str): yield pointer, obj elif isinstance(obj, dict): for k, v in obj.items(): yield from _walk_strings(v, f"{pointer}.{k}") elif isinstance(obj, list): for i, v in enumerate(obj): yield from _walk_strings(v, f"{pointer}[{i}]") def _collect_artifact(*, path: str, obj: Any, ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec, fixture_tags: list[str]) -> list[dict]: out: list[dict] = [] body_hash = _sha256_hex( json.dumps(obj, ensure_ascii=False, sort_keys=True, default=str), ) for ptr, value in _walk_strings(obj): # Each string leaf is scanned via text rules. text = value if not text: continue text_hash = _sha256_hex(text) for rule in _iter_text_rules(ruleset, "existing_artifact"): etype = rule["evidence_type"] rule_id = rule["rule_id"] confidence_band = rule.get("confidence_band_default", "inferred_candidate") for term in rule.get("match_terms", []): for idx in _iter_term_positions(text, term): quote = _extract_sentence_window(text, idx, len(term)) normalized = _normalize_quote(quote) span = { "kind": "artifact_pointer", "char_start": None, "char_end": None, "row_id": None, "row_field": None, "artifact_path": path, "artifact_pointer": ptr, "source_hash": body_hash, } out.append({ "evidence_id": _make_evidence_id( source_kind="existing_artifact", source_ref=f"artifact:{path}", source_span=span, evidence_type=etype, normalized_quote=normalized, ), "source_kind": "existing_artifact", "source_ref": f"artifact:{path}", "source_span": span, "quote": quote, "normalized_quote": normalized, "evidence_type": etype, "candidate_contract_targets": list( _CONTRACT_TARGETS_BY_TYPE.get(etype, []), ), "confidence_band": confidence_band, "match_terms": [term], # plan §2-C: generic basis only — term/pointer 는 # match_terms 와 source_span 에 보존. "inference_basis": ( f"literal artifact field matched by rule {rule_id}" if confidence_band != "observed" else "" ), "sample_fixture_tags": list(fixture_tags), "extracted_by": rule_id, }) # JSON-pointer pull rule (sub_spaces[*].name → space_hint). if isinstance(obj, dict): sub_spaces = obj.get("sub_spaces") or [] if isinstance(sub_spaces, list): for i, sub in enumerate(sub_spaces): name = (sub or {}).get("name") if isinstance(sub, dict) else None if not isinstance(name, str) or not name.strip(): continue quote = name.strip() normalized = _normalize_quote(quote) ptr = f"$.sub_spaces[{i}].name" span = { "kind": "artifact_pointer", "char_start": None, "char_end": None, "row_id": None, "row_field": None, "artifact_path": path, "artifact_pointer": ptr, "source_hash": body_hash, } out.append({ "evidence_id": _make_evidence_id( source_kind="existing_artifact", source_ref=f"artifact:{path}", source_span=span, evidence_type="space_hint", normalized_quote=normalized, ), "source_kind": "existing_artifact", "source_ref": f"artifact:{path}", "source_span": span, "quote": quote, "normalized_quote": normalized, "evidence_type": "space_hint", "candidate_contract_targets": list( _CONTRACT_TARGETS_BY_TYPE.get("space_hint", []), ), "confidence_band": "inferred_candidate", "match_terms": [], # plan §2-C: generic basis — pointer 는 source_span 에 보존. "inference_basis": ( "json pointer pulled by rule existing_artifact_pull" ), "sample_fixture_tags": list(fixture_tags), "extracted_by": "existing_artifact_pull", }) return _dedup_merge(out) def collect_evidence_from_existing_artifact(*, artifacts: list[tuple[str, Any]], ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: out: list[dict] = [] fixture_tags = [f"fixture:{spec.fixture_id}"] for path, obj in artifacts or []: out.extend(_collect_artifact( path=path, obj=obj, ruleset=ruleset, lexicon=lexicon, spec=spec, fixture_tags=fixture_tags, )) return _dedup_merge(out) def collect_evidence_from_diagnostic_run(*, artifacts: list[tuple[str, Any]], ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: """Diagnostic-only path — emits source_kind=existing_artifact (additive only per IMPORTANT 4). Same shape as collect_evidence_from_existing_artifact but tagged for run-meta visibility. """ out: list[dict] = [] fixture_tags = [f"fixture:{spec.fixture_id}", "diagnostic_input"] for path, obj in artifacts or []: out.extend(_collect_artifact( path=path, obj=obj, ruleset=ruleset, lexicon=lexicon, spec=spec, fixture_tags=fixture_tags, )) return _dedup_merge(out) # Convenience: collect everything in one pass (used by tests + main). def collect_all_evidence(*, source_pack: SourcePack, ruleset: dict, lexicon: EvidenceLexicon, spec: SampleFixtureSpec) -> list[dict]: rows: list[dict] = [] rows.extend(collect_evidence_from_planning_doc( text=source_pack.planning_doc_text, ruleset=ruleset, lexicon=lexicon, spec=spec, )) rows.extend(collect_evidence_from_episode_fulltext( text=source_pack.episode_fulltext, ruleset=ruleset, lexicon=lexicon, spec=spec, )) rows.extend(collect_evidence_from_shots( shots=source_pack.selected_shots, ruleset=ruleset, lexicon=lexicon, spec=spec, )) rows.extend(collect_evidence_from_entity_catalog( entities=source_pack.entity_catalog, ruleset=ruleset, lexicon=lexicon, spec=spec, )) rows.extend(collect_evidence_from_location_catalog( entities=source_pack.location_catalog, ruleset=ruleset, lexicon=lexicon, spec=spec, )) rows.extend(collect_evidence_from_existing_artifact( artifacts=source_pack.existing_artifacts, ruleset=ruleset, lexicon=lexicon, spec=spec, )) rows.extend(collect_evidence_from_diagnostic_run( artifacts=source_pack.diagnostic_artifacts, ruleset=ruleset, lexicon=lexicon, spec=spec, )) return _dedup_merge(rows) # ============================================================================ # W1c review helpers — noisy heuristic + readiness dimensions (plan §8-E) # ============================================================================ def _is_noisy_candidate(*, quote: str, match_terms: list[str], negation_terms: list[str]) -> bool: """True iff `quote` contains both a negation marker AND at least one of the row's `match_terms`. structural rows (empty match_terms) are NOT flagged. English terms expressed as `\\b...\\b` regex literals get word- boundary matching to avoid `know` ⊃ `no` false positives. """ if not quote or not match_terms: return False q = quote has_negation = False for nt in negation_terms or []: if not nt: continue # Heuristic: a token starting with `\b` is a regex (English path). if nt.startswith("\\b") or nt.startswith("\\B") or nt.startswith("("): try: if re.search(nt, q, flags=re.IGNORECASE): has_negation = True break except re.error: continue else: # Plain substring (Korean phrase). bare `안 ` is banned upstream so # the only risk is the caller adding it manually — we trust the # lexicon here. if nt in q: has_negation = True break if not has_negation: # Fast path also try the precompiled English union (handles fixture # that strips negation_terms but still wants generic English guard). if _ENGLISH_NEGATION_PATTERN.search(q): has_negation = True if not has_negation: return False for term in match_terms: if term and term in q: return True return False def _evaluate_readiness(*, evidence_rows: list[dict], spec: SampleFixtureSpec) -> dict[str, dict[str, Any]]: """Per-dimension readiness report. Returns dict keyed by dimension — `topology_candidate / state_model / camera_intent / overall` — each value is `{status, observed, threshold, missing}` where status ∈ {yes, conditional, weak, no}. Thresholds come from `spec.readiness_thresholds` only; no hardcoded values in this function body (plan §8-E IMPORTANT). `status` derivation: - all metric thresholds met → "yes" - some met (≥1 short of full set) → "conditional" - none met → "no"; "weak" used for very-low coverage (camera 1-term 등) - overall = worst dimension (yes → conditional → weak → no priority). """ thresholds = spec.readiness_thresholds # Distinct term tallies per evidence_type. distinct_terms_by_type: dict[str, set[str]] = {} row_count_by_type: dict[str, int] = {} state_rows: list[dict] = [] noisy_state_rows: list[dict] = [] for r in evidence_rows: et = r["evidence_type"] row_count_by_type[et] = row_count_by_type.get(et, 0) + 1 for t in r.get("match_terms") or []: distinct_terms_by_type.setdefault(et, set()).add(t) if et == "state_hint": state_rows.append(r) if _is_noisy_candidate( quote=r.get("quote") or "", match_terms=r.get("match_terms") or [], negation_terms=spec.negation_terms): noisy_state_rows.append(r) def _dim_status(observed: dict[str, Any], threshold: dict[str, Any], weak_predicate=None) -> tuple[str, list[str]]: missing: list[str] = [] for k, v in threshold.items(): o = observed.get(k, 0) if k == "max_noisy_ratio": # Lower-is-better metric. if o > v: missing.append(f"{k}: {o:.3f} > {v}") else: if o < v: missing.append(f"{k}: {o} < {v}") if not missing: return "yes", [] # "weak" reserved for camera_intent very-low coverage. if weak_predicate and weak_predicate(observed): return "weak", missing if len(missing) < len(threshold): return "conditional", missing return "no", missing # topology_candidate — place/space use distinct match-terms, boundary/ # door-window use row counts (plan §8-F W1d IMPORTANT 2 key rename). topo_obs = { "place_distinct_terms": len(distinct_terms_by_type.get("place_hint", set())), "space_distinct_terms": len(distinct_terms_by_type.get("space_hint", set())), "boundary_rows": row_count_by_type.get("boundary_hint", 0), "door_window_rows": row_count_by_type.get("door_window_hint", 0), } topo_status, topo_missing = _dim_status( topo_obs, thresholds["topology_candidate"], ) # state_model — coverage 통과 + noisy_rows>0 시 conditional 다운그레이드 # (plan §8-F W1d IMPORTANT 1). yes 는 coverage 통과 AND noisy=0 일 때만. total_state = max(len(state_rows), 1) noisy_ratio = len(noisy_state_rows) / total_state state_obs = { "state_distinct_terms": len(distinct_terms_by_type.get("state_hint", set())), "max_noisy_ratio": noisy_ratio, } state_status, state_missing = _dim_status( state_obs, thresholds["state_model"], ) if state_status == "yes" and len(noisy_state_rows) > 0: state_status = "conditional" state_missing = list(state_missing) + [ f"state_noisy_rows: {len(noisy_state_rows)} rows require review " f"(noisy_ratio={noisy_ratio:.3f}, threshold " f"max_noisy_ratio<{thresholds['state_model']['max_noisy_ratio']})" ] # camera_intent — "weak" applies when camera distinct = 1 AND movement = 0. cam_obs = { "camera_distinct_terms": len(distinct_terms_by_type.get("camera_hint", set())), "movement_rows": row_count_by_type.get("movement_hint", 0), } cam_status, cam_missing = _dim_status( cam_obs, thresholds["camera_intent"], weak_predicate=lambda obs: ( obs["camera_distinct_terms"] <= 1 and obs["movement_rows"] == 0 ), ) # Overall = worst dimension. Priority order yes > conditional > weak > no. priority = {"yes": 0, "conditional": 1, "weak": 2, "no": 3} worst = max((topo_status, state_status, cam_status), key=lambda s: priority.get(s, 9)) overall_missing: list[str] = [] for label, st, ms in [ ("topology_candidate", topo_status, topo_missing), ("state_model", state_status, state_missing), ("camera_intent", cam_status, cam_missing), ]: if st != "yes" and ms: overall_missing.extend([f"{label}/{m}" for m in ms]) return { "topology_candidate": { "status": topo_status, "observed": topo_obs, "threshold": thresholds["topology_candidate"], "missing": topo_missing, }, "state_model": { "status": state_status, "observed": state_obs, "threshold": thresholds["state_model"], "missing": state_missing, "state_total_rows": len(state_rows), "state_noisy_rows": len(noisy_state_rows), }, "camera_intent": { "status": cam_status, "observed": cam_obs, "threshold": thresholds["camera_intent"], "missing": cam_missing, }, "overall": { "status": worst, "missing": overall_missing, }, } # ============================================================================ # Coverage summary + acceptance metric # ============================================================================ def _generic_function_source_for_leakage() -> str: """Concatenate the source of all generic collectors / loaders so that runtime leakage guard can re-confirm the static guard. AST not required at runtime — string search is sufficient for the FORBIDDEN_LEAKAGE_LITERALS check. """ import inspect as _inspect targets: list[str] = [] for name in dir(sys.modules[__name__]): if not name.startswith( ("collect_", "load_episode_fulltext", "load_planning_doc", "load_selected_shots", "load_entity_catalog", "load_location_catalog", "build_evidence_extraction_ruleset"), ): continue obj = getattr(sys.modules[__name__], name) if callable(obj): try: targets.append(_inspect.getsource(obj)) except (OSError, TypeError): continue return "\n".join(targets) def _runtime_leakage_clean() -> bool: body = _generic_function_source_for_leakage() for lit in FORBIDDEN_LEAKAGE_LITERALS_FOR_RUNTIME: if lit in body: return False return True # Runtime-only leakage list — mirrors the test guard. Defined inline so that # the script can self-check during coverage build. FORBIDDEN_LEAKAGE_LITERALS_FOR_RUNTIME = [ "L05", "옥탑방", "수리영", "민숙", "안방", "pg_rooftop_villa", "sg_rooftop_interior", ] _REQUIRED_SOURCE_KINDS_STRICT = [ "planning_doc", "episode_fulltext", "shot_description", "entity_catalog", "location_catalog", ] _REQUIRED_EVIDENCE_TYPES_STRICT = [ "place_hint", "space_hint", "state_hint", "camera_hint", ] def build_source_coverage_summary(*, evidence_rows: list[dict], spec: SampleFixtureSpec, acceptance_mode: str) -> dict: """Aggregate counts + run acceptance gate. plan §6 acceptance: strict on sample_fixture, advisory on override (IMPORTANT 3). """ if acceptance_mode not in {"sample_fixture", "override"}: raise ValueError( f"acceptance_mode must be sample_fixture|override, " f"got {acceptance_mode!r}" ) source_kind_counts: dict[str, int] = {} evidence_type_counts: dict[str, int] = {} confidence_band_counts: dict[str, int] = {} ambiguous_rows: list[dict] = [] inference_basis_missing: list[str] = [] match_terms_missing: list[str] = [] for r in evidence_rows: source_kind_counts[r["source_kind"]] = ( source_kind_counts.get(r["source_kind"], 0) + 1 ) evidence_type_counts[r["evidence_type"]] = ( evidence_type_counts.get(r["evidence_type"], 0) + 1 ) confidence_band_counts[r["confidence_band"]] = ( confidence_band_counts.get(r["confidence_band"], 0) + 1 ) if r["confidence_band"] != "observed": if not r.get("inference_basis"): inference_basis_missing.append(r["evidence_id"]) if not r.get("match_terms"): match_terms_missing.append(r["evidence_id"]) if (r["evidence_type"] == "ambiguity_hint" or r["confidence_band"] == "ambiguous"): ambiguous_rows.append({ "evidence_id": r["evidence_id"], "source_kind": r["source_kind"], "source_ref": r["source_ref"], "quote": r["quote"], "evidence_type": r["evidence_type"], }) leakage_clean = _runtime_leakage_clean() # Acceptance gate ----------------------------------------------------- advisory_notes: list[str] = [] hard_pass = True hard_fails: list[str] = [] # Count threshold + source_kind / evidence_type coverage. if acceptance_mode == "sample_fixture": if len(evidence_rows) < 50: hard_pass = False hard_fails.append( f"evidence_count={len(evidence_rows)} < 50 (sample fixture " f"requires ≥50)" ) for kind in _REQUIRED_SOURCE_KINDS_STRICT: if source_kind_counts.get(kind, 0) < 1: hard_pass = False hard_fails.append(f"source_kind {kind!r} missing") type_hits = sum( 1 for etype in _REQUIRED_EVIDENCE_TYPES_STRICT if evidence_type_counts.get(etype, 0) >= 1 ) if type_hits < 4: hard_pass = False hard_fails.append( f"evidence_type coverage {type_hits}/4 of " f"{_REQUIRED_EVIDENCE_TYPES_STRICT}" ) else: # override — advisory only. if len(evidence_rows) < 50: advisory_notes.append( f"advisory: evidence_count={len(evidence_rows)} below 50" ) for kind in _REQUIRED_SOURCE_KINDS_STRICT: if source_kind_counts.get(kind, 0) < 1: advisory_notes.append( f"advisory: source_kind {kind!r} missing" ) # Schema invariants — always hard, even in override mode. if inference_basis_missing: hard_pass = False hard_fails.append( f"non-observed rows missing inference_basis: " f"{len(inference_basis_missing)}" ) if match_terms_missing: hard_pass = False hard_fails.append( f"non-observed rows missing match_terms: {len(match_terms_missing)}" ) if not leakage_clean: hard_pass = False hard_fails.append( "FORBIDDEN_LEAKAGE_LITERALS detected in generic function bodies" ) return { "acceptance_mode": acceptance_mode, "fixture_id": spec.fixture_id, "evidence_count": len(evidence_rows), "source_kind_counts": source_kind_counts, "evidence_type_counts": evidence_type_counts, "confidence_band_counts": confidence_band_counts, "leakage_guard_clean": leakage_clean, "ambiguous_rows": ambiguous_rows[:50], "ambiguous_rows_truncated": len(ambiguous_rows) > 50, "acceptance": { "hard_pass": hard_pass, "hard_fails": hard_fails, }, "advisory_notes": advisory_notes, } # ============================================================================ # Input manifest + run meta # ============================================================================ def build_input_manifest(*, spec: SampleFixtureSpec, source_pack: SourcePack, ruleset: dict, args_dict: dict) -> dict: sources: dict[str, dict] = {} sources["planning_doc"] = { "ref": f"project_registry.planning_doc_text where id={spec.project_id}", "sha256": _sha256_hex(source_pack.planning_doc_text or ""), "length_chars": len(source_pack.planning_doc_text or ""), } sources["episode_fulltext"] = { "ref": f"episode.fulltext where id={spec.episode_id}", "sha256": _sha256_hex(source_pack.episode_fulltext or ""), "length_chars": len(source_pack.episode_fulltext or ""), } shots_dump = json.dumps(source_pack.selected_shots, ensure_ascii=False, sort_keys=True, default=str) sources["selected_shots"] = { "ref": ( f"scene_still where episode_id={spec.episode_id} AND " f"is_selected=True" ), "sha256": _sha256_hex(shots_dump), "row_count": len(source_pack.selected_shots), } ent_dump = json.dumps(source_pack.entity_catalog, ensure_ascii=False, sort_keys=True, default=str) sources["entity_catalog"] = { "ref": ( f"entity_canon where project_id={spec.project_id} AND " f"entity_type IN (character,prop,outlook)" ), "sha256": _sha256_hex(ent_dump), "row_count": len(source_pack.entity_catalog), } loc_dump = json.dumps(source_pack.location_catalog, ensure_ascii=False, sort_keys=True, default=str) sources["location_catalog"] = { "ref": ( f"entity_canon where project_id={spec.project_id} AND " f"entity_type='location'" ), "sha256": _sha256_hex(loc_dump), "row_count": len(source_pack.location_catalog), } if source_pack.existing_artifacts: artifact_dump = json.dumps( [(p, o) for p, o in source_pack.existing_artifacts], ensure_ascii=False, sort_keys=True, default=str, ) sources["existing_artifact"] = { "ref": str(spec.source_run_path), "sha256": _sha256_hex(artifact_dump), "row_count": len(source_pack.existing_artifacts), } if source_pack.diagnostic_artifacts: diag_dump = json.dumps( [(p, o) for p, o in source_pack.diagnostic_artifacts], ensure_ascii=False, sort_keys=True, default=str, ) sources["diagnostic_artifact"] = { "ref": "diagnostic runs (additive only)", "sha256": _sha256_hex(diag_dump), "row_count": len(source_pack.diagnostic_artifacts), } return { "fixture_id": spec.fixture_id, "project_id": spec.project_id, "episode_id": spec.episode_id, "plan_version": PLAN_VERSION, "ruleset_id": ruleset.get("ruleset_id"), "lexicon_hash": ruleset.get("lexicon_hash"), "missing_inputs": list(source_pack.missing_inputs), "args": args_dict, "sources": sources, } def _now_iso() -> str: # KST timezone for run_meta clarity. return datetime.now(timezone(timedelta(hours=9))).isoformat(timespec="seconds") def _short_uuid() -> str: return uuid.uuid4().hex[:6] def _run_id() -> str: return f"{datetime.now(timezone(timedelta(hours=9))).strftime('%Y%m%d_%H%M')}_{_short_uuid()}" def build_run_meta(*, run_id: str, fixture_id: str, args_dict: dict, outputs: list[str], lexicon_hash: str) -> dict: return { "run_id": run_id, "plan_version": PLAN_VERSION, "generated_at": _now_iso(), "fixture_id": fixture_id, "args": args_dict, "outputs": list(outputs), "lexicon_hash": lexicon_hash, } # ============================================================================ # HTML renderer # ============================================================================ def _esc(value: Any) -> str: return html_lib.escape(str(value), quote=True) def _status_badge_class(status: str) -> str: return { "yes": "ok", "conditional": "warn", "weak": "warn", "no": "bad", }.get(status, "warn") def _render_readiness_summary(readiness: dict[str, dict[str, Any]]) -> str: overall = readiness.get("overall", {}) overall_status = overall.get("status", "no") parts: list[str] = [] parts.append("

§0. W2 readiness summary

") parts.append( f"

overall = " f"{_esc(overall_status)}

" ) parts.append("" "" "") for dim in ("topology_candidate", "state_model", "camera_intent"): d = readiness.get(dim, {}) cls = _status_badge_class(d.get("status", "no")) parts.append( f"" f"" f"" f"" f"" ) parts.append("
dimensionstatusobservedthresholdmissing
{_esc(dim)}{_esc(d.get('status'))}{_esc(d.get('observed'))}{_esc(d.get('threshold'))}{'
'.join(_esc(m) for m in d.get('missing', []))}" f"
") parts.append( "

⚠ ambiguity classification 미실행 — " "W1 collector 는 literal-term matching 만 수행합니다. ambiguous=0 은 " "'모호 근거 없음' 을 뜻하지 않습니다 — §6 noisy candidates 섹션 참조.

" ) return "".join(parts) def _render_grouped_evidence(*, evidence_rows: list[dict], etypes: list[str], heading: str, max_quotes_per_group: int = 5) -> str: """Group evidence rows by (evidence_type, primary match_term) and render quote excerpts per group. Used for §3 Place/Set/Space, §4 State. """ parts: list[str] = [] parts.append(f"

{heading}

") groups: dict[tuple[str, str], list[dict]] = {} for r in evidence_rows: if r["evidence_type"] not in etypes: continue terms = r.get("match_terms") or [""] # primary term = first sorted, or "(structural)" when empty. primary = (sorted(terms)[0] if terms and terms[0] else "(structural)") groups.setdefault((r["evidence_type"], primary), []).append(r) if not groups: parts.append("

(no rows)

") return "".join(parts) parts.append("" "" "") for (etype, term), rows in sorted(groups.items()): quotes = [r.get("quote", "") for r in rows[:max_quotes_per_group]] parts.append( f"" f"" f"" f"" ) parts.append("
evidence_typecandidaterow countsample quotes
{_esc(etype)}{_esc(term)}{len(rows)}{'
'.join(_esc(q) for q in quotes)}
") return "".join(parts) def _render_camera_evidence(evidence_rows: list[dict]) -> str: parts: list[str] = [] parts.append("

§5. Camera evidence (per-shot)

") by_shot: dict[str, list[dict]] = {} for r in evidence_rows: if r["evidence_type"] not in {"camera_hint", "movement_hint"}: continue by_shot.setdefault(r["source_ref"], []).append(r) if not by_shot: parts.append("

(no camera/movement evidence)

") return "".join(parts) parts.append("" "" "") for ref in sorted(by_shot.keys()): for r in by_shot[ref][:5]: parts.append( f"" f"" f"" f"" ) parts.append("
source_ref (shot)evidence_typematch_termsquote
{_esc(ref)}{_esc(r['evidence_type'])}{_esc(','.join(r.get('match_terms') or []))}{_esc(r.get('quote', ''))}
") return "".join(parts) def _render_noisy_candidates(*, evidence_rows: list[dict], spec: SampleFixtureSpec, limit: int = 50) -> str: parts: list[str] = [] parts.append("

§6. Noisy evidence candidates " "(negation marker + match_term in quote)

") noisy: list[dict] = [] for r in evidence_rows: if _is_noisy_candidate( quote=r.get("quote") or "", match_terms=r.get("match_terms") or [], negation_terms=spec.negation_terms): noisy.append(r) if not noisy: parts.append("

(no noisy candidates flagged)

") return "".join(parts) parts.append(f"

total flagged = {len(noisy)}" f" (showing first {min(limit, len(noisy))})

") parts.append("" "" "") for r in noisy[:limit]: parts.append( f"" f"" f"" f"" f"" ) parts.append("
evidence_idtypesourcematch_termsquote
{_esc(r['evidence_id'])}{_esc(r['evidence_type'])}{_esc(r['source_kind'])}{_esc(','.join(r.get('match_terms') or []))}{_esc(r.get('quote', ''))}
") return "".join(parts) def _render_top_source_quotes(*, evidence_rows: list[dict], per_kind: int = 5) -> str: parts: list[str] = [] parts.append("

§7. Top source quotes (per source_kind)

") by_kind: dict[str, list[dict]] = {} for r in evidence_rows: by_kind.setdefault(r["source_kind"], []).append(r) if not by_kind: parts.append("

(no evidence)

") return "".join(parts) for kind in sorted(by_kind.keys()): rows = by_kind[kind] # Prefer diversity: max one row per evidence_type, longest quote first. seen_etypes: set[str] = set() picks: list[dict] = [] for r in sorted(rows, key=lambda r: -len(r.get("quote") or "")): et = r["evidence_type"] if et in seen_etypes: continue seen_etypes.add(et) picks.append(r) if len(picks) >= per_kind: break if not picks: picks = rows[:per_kind] parts.append(f"

{_esc(kind)} ({len(rows)} rows)

") parts.append("" "") for r in picks: parts.append( f"" f"" f"" ) parts.append("
evidence_typematch_termsquote
{_esc(r['evidence_type'])}{_esc(','.join(r.get('match_terms') or []))}{_esc(r.get('quote', ''))}
") return "".join(parts) def _render_detailed_readiness(readiness: dict[str, dict[str, Any]]) -> str: parts: list[str] = [] parts.append("

§8. Detailed readiness (per-dimension)

") for dim in ("topology_candidate", "state_model", "camera_intent"): d = readiness.get(dim, {}) cls = _status_badge_class(d.get("status", "no")) parts.append(f"

{_esc(dim)} — " f"{_esc(d.get('status'))}

") parts.append("
    ") for k, v in (d.get("observed") or {}).items(): t = (d.get("threshold") or {}).get(k) parts.append( f"
  • {_esc(k)}: observed={_esc(v)} / " f"threshold={_esc(t)}
    • " ) parts.append("
") if d.get("missing"): parts.append("

missing:

    ") for m in d["missing"]: parts.append(f"
  • {_esc(m)}
    • ") parts.append("
") return "".join(parts) def render_html(*, evidence_rows: list[dict], ruleset: dict, coverage: dict, spec: SampleFixtureSpec, run_meta: dict) -> str: """W1c review-cockpit HTML (plan §8-E). 첫 화면 = W2 readiness summary + ambiguity 경고; 그 다음 흐름/counts/evidence sections/noisy/top quotes/ detailed readiness. raw evidence pack 은 변경 0 — readiness 와 noisy 는 render 시 동적 계산. """ readiness = _evaluate_readiness( evidence_rows=evidence_rows, spec=spec, ) parts: list[str] = [] parts.append( "" "Background Topology Planner — W1c review cockpit" "" ) parts.append( f"

Background Topology Planner — generic experiment review " f"cockpit (sample fixture: {_esc(spec.fixture_id)})

" ) parts.append( f"

review/report enhancement" f" run_id={_esc(run_meta.get('run_id'))}" f" / plan_version={_esc(run_meta.get('plan_version'))}" f" / acceptance_mode={_esc(coverage.get('acceptance_mode'))}" f" / lexicon_hash={_esc(run_meta.get('lexicon_hash'))[:16]}…

" ) source_run = run_meta.get("source_run_dir") or run_meta.get("source_run") if source_run: parts.append( f"

source_run_dir = {_esc(source_run)}

" ) # §0 — W2 readiness summary (TOP of page per plan §8-E). parts.append(_render_readiness_summary(readiness)) # §1 — flow. parts.append( "

§1. 흐름 요약

    " "
  1. source evidence → topology candidate → shot spatial intent →" " background unit need → spatial decision ledger
    2. " "
  2. 현재 wave = W1c = review cockpit over approved W1b " "snapshot (no re-extraction)
    3. " "
  3. next: W2 = topology candidate generation" " (SetTopologyGraph + StructuralStateModel)
    4. " "
" ) # §2 — counts 축약. parts.append("

§2. source_kind 별 evidence count

") parts.append("") sk = coverage.get("source_kind_counts", {}) for kind in ( "planning_doc", "episode_fulltext", "shot_description", "scene_summary", "entity_catalog", "location_catalog", "existing_artifact", ): parts.append( f"" f"" ) parts.append("
source_kindcount
{_esc(kind)}{sk.get(kind, 0)}
") parts.append("

evidence_type / confidence_band 분포는 " "source_coverage_summary.json 참조.

") # §3 — Place / Set / Space evidence. parts.append(_render_grouped_evidence( evidence_rows=evidence_rows, etypes=["place_hint", "set_hint", "space_hint", "zone_hint", "boundary_hint", "door_window_hint", "furniture_hint"], heading="§3. Place / Set / Space evidence (candidate별 quote)", )) # §4 — State evidence (group by match_term + noisy flag inline). parts.append("

§4. State evidence (match_term group + noisy flag)

") state_groups: dict[str, list[dict]] = {} for r in evidence_rows: if r["evidence_type"] != "state_hint": continue terms = r.get("match_terms") or ["(structural)"] primary = (sorted(terms)[0] if terms and terms[0] else "(structural)") state_groups.setdefault(primary, []).append(r) if not state_groups: parts.append("

(no state evidence)

") else: parts.append("" "" "") for term in sorted(state_groups.keys()): rows = state_groups[term] noisy_count = sum( 1 for r in rows if _is_noisy_candidate( quote=r.get("quote") or "", match_terms=r.get("match_terms") or [], negation_terms=spec.negation_terms, ) ) quotes = [r.get("quote", "") for r in rows[:3]] parts.append( f"" f"" f"" f"" ) parts.append("
match_termtotalnoisysample quotes
{_esc(term)}{len(rows)}" f"{noisy_count}{'
'.join(_esc(q) for q in quotes)}
") # §5 — Camera evidence per shot. parts.append(_render_camera_evidence(evidence_rows)) # §6 — Noisy evidence candidates. parts.append(_render_noisy_candidates( evidence_rows=evidence_rows, spec=spec, )) # §7 — Top source quotes. parts.append(_render_top_source_quotes(evidence_rows=evidence_rows)) # §8 — Detailed readiness. parts.append(_render_detailed_readiness(readiness)) # Leakage / acceptance footer (compact). leakage_ok = coverage.get("leakage_guard_clean", False) leakage_cls = "ok" if leakage_ok else "bad" accept = coverage.get("acceptance", {}) parts.append( f"

Footer — leakage / acceptance

" f"

" f"FORBIDDEN_LEAKAGE_LITERALS check: " f"{'CLEAN' if leakage_ok else 'LEAKED'}" f" / acceptance.hard_pass={accept.get('hard_pass')}" f"

" ) if accept.get("hard_fails"): parts.append("
    ") for f in accept["hard_fails"]: parts.append(f"
  • {_esc(f)}
    • ") parts.append("
") parts.append("") return "".join(parts) # ============================================================================ # DB readers (generic, sibling-experiment-free) # ============================================================================ def load_episode_fulltext(session, episode_id: str) -> tuple[str, dict]: """Return (fulltext, meta). meta carries id/title/length.""" from app.models.project import Episode ep = session.query(Episode).filter(Episode.id == episode_id).one_or_none() if ep is None: return "", {"missing": True, "id": episode_id} fulltext = (ep.fulltext or "").strip() return fulltext, { "id": ep.id, "episode_number": ep.episode_number, "title": ep.title, "length_chars": len(fulltext), } def load_planning_doc(session, project_id: str) -> tuple[str, dict]: from app.models.catalog import ProjectRegistry proj = session.query(ProjectRegistry).filter( ProjectRegistry.id == project_id, ).one_or_none() if proj is None: return "", {"missing": True, "id": project_id} text = (proj.planning_doc_text or "").strip() return text, { "id": proj.id, "name": proj.name, "length_chars": len(text), } def load_selected_shots(session, project_id: str, episode_id: str, loc_short_ids: Optional[list[str]] = None) -> list[dict]: """Generic SceneStill loader. plan §2-A BLOCKING 3 — sibling experiment helpers (load_l05_shots / ShotMeta) replaced by this in-script reader. Filter rules: - episode_id == episode_id - is_selected == True - still_index >= 0 - status IS NULL OR status != 'stale' - (optional) visible_entities_json substring match against any loc_short_ids entry — caller responsibility, simple filter only. """ from app.models.project import SceneStill # plan §8-D: SQL-level deterministic order_by + Python fallback sort so # cross-run row order is stable regardless of dialect / index plan. q = session.query(SceneStill).filter( SceneStill.episode_id == episode_id, SceneStill.project_id == project_id, SceneStill.is_selected.is_(True), SceneStill.still_index >= 0, ).order_by( SceneStill.scene_index, SceneStill.shot_index, SceneStill.still_index, SceneStill.id, ) rows = q.all() out: list[dict] = [] for r in rows: status = getattr(r, "status", None) if status == "stale": continue ve = getattr(r, "visible_entities_json", None) or "[]" if loc_short_ids: if not any(short in ve for short in loc_short_ids): continue out.append({ "still_id": r.id, "scene_index": getattr(r, "scene_index", None), "shot_index": getattr(r, "shot_index", None), "shot_description": getattr(r, "shot_description", "") or "", "scene_summary": getattr(r, "scene_summary", "") or "", "visible_entities_json": ve, "is_selected": True, "status": status, }) out.sort(key=lambda d: ( d.get("scene_index") if d.get("scene_index") is not None else -1, d.get("shot_index") if d.get("shot_index") is not None else -1, str(d.get("still_id") or ""), )) return out def load_entity_catalog(session, project_id: str) -> list[dict]: from app.models.project import EntityCanon # plan §8-D: deterministic order_by (entity_type → short_id → id) + Python # fallback sort. EntityCanon.short_id 는 backend/app/models/project.py:33 # 컬럼. rows = session.query(EntityCanon).filter( EntityCanon.project_id == project_id, EntityCanon.entity_type.in_(["character", "prop", "outlook"]), ).order_by( EntityCanon.entity_type, EntityCanon.short_id, EntityCanon.id, ).all() out: list[dict] = [] for r in rows: out.append({ "canon_id": r.id, "entity_type": r.entity_type, "short_id": getattr(r, "short_id", None), "name": r.name, "description": r.description or "", "metadata_json": r.metadata_json or "{}", }) out.sort(key=lambda d: ( str(d.get("entity_type") or ""), str(d.get("short_id") or ""), str(d.get("canon_id") or ""), )) return out def load_location_catalog(session, project_id: str) -> list[dict]: from app.models.project import EntityCanon # plan §8-D: same deterministic ordering as load_entity_catalog. rows = session.query(EntityCanon).filter( EntityCanon.project_id == project_id, EntityCanon.entity_type == "location", ).order_by( EntityCanon.entity_type, EntityCanon.short_id, EntityCanon.id, ).all() out: list[dict] = [] for r in rows: out.append({ "canon_id": r.id, "entity_type": r.entity_type, "short_id": getattr(r, "short_id", None), "name": r.name, "description": r.description or "", "metadata_json": r.metadata_json or "{}", }) out.sort(key=lambda d: ( str(d.get("entity_type") or ""), str(d.get("short_id") or ""), str(d.get("canon_id") or ""), )) return out # ============================================================================ # write_outputs — produces the 7 files # ============================================================================ def _tsv_columns() -> list[str]: return [ "evidence_id", "source_kind", "source_ref", "span_kind", "char_start", "char_end", "row_id", "row_field", "artifact_path", "artifact_pointer", "source_hash", "evidence_type", "confidence_band", "match_terms", "candidate_contract_targets", "inference_basis", "extracted_by", "sample_fixture_tags", "quote", ] def _row_for_tsv(row: dict) -> dict: span = row.get("source_span") or {} return { "evidence_id": row.get("evidence_id", ""), "source_kind": row.get("source_kind", ""), "source_ref": row.get("source_ref", ""), "span_kind": span.get("kind", ""), "char_start": span.get("char_start", ""), "char_end": span.get("char_end", ""), "row_id": span.get("row_id", ""), "row_field": span.get("row_field", ""), "artifact_path": span.get("artifact_path", ""), "artifact_pointer": span.get("artifact_pointer", ""), "source_hash": span.get("source_hash", ""), "evidence_type": row.get("evidence_type", ""), "confidence_band": row.get("confidence_band", ""), "match_terms": ",".join(row.get("match_terms", []) or []), "candidate_contract_targets": ",".join( row.get("candidate_contract_targets", []) or [], ), "inference_basis": row.get("inference_basis", ""), "extracted_by": ",".join( row.get("extracted_by", []) if isinstance( row.get("extracted_by"), list, ) else [row.get("extracted_by") or ""], ), "sample_fixture_tags": ",".join( row.get("sample_fixture_tags", []) or [], ), "quote": (row.get("quote") or "").replace("\t", " ").replace("\n", " "), } def write_outputs(*, run_dir: Path, evidence_rows: list[dict], ruleset: dict, input_manifest: dict, coverage_summary: dict, html: str, run_meta: dict) -> None: run_dir.mkdir(parents=True, exist_ok=True) # plan §8-D: sort by evidence_id for cross-run byte stability across # source_evidence_pack.json + source_evidence.tsv. evidence_id is a # deterministic sha256 prefix so this order is stable and reproducible. evidence_rows = sorted( evidence_rows, key=lambda r: str(r.get("evidence_id") or ""), ) # 1) input_manifest.json (run_dir / "input_manifest.json").write_text( json.dumps(input_manifest, ensure_ascii=False, indent=2), encoding="utf-8", ) # 2) source_evidence_pack.json (run_dir / "source_evidence_pack.json").write_text( json.dumps({"rows": evidence_rows}, ensure_ascii=False, indent=2), encoding="utf-8", ) # 3) source_evidence.tsv tsv_path = run_dir / "source_evidence.tsv" with tsv_path.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=_tsv_columns(), delimiter="\t") writer.writeheader() for r in evidence_rows: writer.writerow(_row_for_tsv(r)) # 4) evidence_extraction_ruleset.json (run_dir / "evidence_extraction_ruleset.json").write_text( json.dumps(ruleset, ensure_ascii=False, indent=2), encoding="utf-8", ) # 5) source_coverage_summary.json (run_dir / "source_coverage_summary.json").write_text( json.dumps(coverage_summary, ensure_ascii=False, indent=2), encoding="utf-8", ) # 6) index.html (run_dir / "index.html").write_text(html, encoding="utf-8") # 7) run_meta.json (run_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8", ) # ============================================================================ # Diagnostic run discovery (read-only) # ============================================================================ _DIAGNOSTIC_RUN_PARENTS = [ Path("scripts_output/background_place_grouping_experiment"), Path("scripts_output/background_spatial_decision_experiment"), ] _DIAGNOSTIC_FILES = [ "place_groups.json", "set_groups.json", "space_nodes.json", "state_layers.json", "structural_versions.json", "shot_bindings.json", "chain_bg_decomposition.json", "generation_unit_plan.json", "shot_spatial_decisions.json", "rollup_requests.json", ] def _resolve_latest_run(parent: Path) -> Optional[Path]: if not parent.exists() or not parent.is_dir(): return None candidates = sorted([ p for p in parent.iterdir() if p.is_dir() and re.match(r"^\d{8}_\d{4}_[0-9a-f]+$", p.name) ]) return candidates[-1] if candidates else None _W1B_RAW_FILES: list[str] = [ "source_evidence_pack.json", "source_evidence.tsv", "evidence_extraction_ruleset.json", "input_manifest.json", "source_coverage_summary.json", ] def run_from_existing_run(*, source_run_dir: Path, dest_run_dir: Path) -> dict[str, Any]: """W1c review-cockpit mode (plan §8-E). Copies 5 raw artifacts from `source_run_dir` byte-identical into `dest_run_dir`, then emits new `index.html`, `readiness_report.json`, and `run_meta.json` carrying W1c metadata. raw evidence/ruleset/manifest/coverage 변경 0. Returns the run_meta dict written into dest_run_dir/run_meta.json. """ source_run_dir = Path(source_run_dir) dest_run_dir = Path(dest_run_dir) dest_run_dir.mkdir(parents=True, exist_ok=True) # 1) Copy raw 5 files byte-identical. for fname in _W1B_RAW_FILES: src_path = source_run_dir / fname if not src_path.exists(): raise FileNotFoundError( f"source_run_dir missing required raw file {fname!r}: " f"{src_path}" ) dest_run_dir.joinpath(fname).write_bytes(src_path.read_bytes()) # 2) Load evidence pack + coverage + ruleset (read-only). pack = json.loads( (dest_run_dir / "source_evidence_pack.json").read_text(encoding="utf-8"), ) evidence_rows = pack.get("rows", []) coverage = json.loads( (dest_run_dir / "source_coverage_summary.json").read_text(encoding="utf-8"), ) ruleset = json.loads( (dest_run_dir / "evidence_extraction_ruleset.json").read_text(encoding="utf-8"), ) # 3) Reuse build_sample_fixture_l05_spec — readiness_thresholds / # negation_terms 가 dataclass default 로 자동 적용. spec = build_sample_fixture_l05_spec() readiness = _evaluate_readiness( evidence_rows=evidence_rows, spec=spec, ) # 4) Compose run_meta — plan_version='btp_w1c' + source_run_dir reference. new_run_id = dest_run_dir.name or _run_id() run_meta = { "run_id": new_run_id, "plan_version": "btp_w1c", "generated_at": _now_iso(), "fixture_id": spec.fixture_id, "args": { "mode": "from_run_dir", "source_run_dir": str(source_run_dir), }, "outputs": _W1B_RAW_FILES + [ "index.html", "readiness_report.json", "run_meta.json", ], "lexicon_hash": ruleset.get("lexicon_hash"), "source_run_dir": str(source_run_dir), } # 5) Emit new index.html (review cockpit) + readiness_report.json. html = render_html( evidence_rows=evidence_rows, ruleset=ruleset, coverage=coverage, spec=spec, run_meta=run_meta, ) (dest_run_dir / "index.html").write_text(html, encoding="utf-8") (dest_run_dir / "readiness_report.json").write_text( json.dumps(readiness, ensure_ascii=False, indent=2), encoding="utf-8", ) (dest_run_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8", ) return run_meta def load_diagnostic_artifacts(repo_root: Path) -> list[tuple[str, Any]]: """Load latest diagnostic run JSONs from sibling experiment output dirs. Returns list of (relative_path_string, parsed_obj). Read-only. """ out: list[tuple[str, Any]] = [] for parent_rel in _DIAGNOSTIC_RUN_PARENTS: parent_abs = repo_root / parent_rel latest = _resolve_latest_run(parent_abs) if latest is None: continue for fname in _DIAGNOSTIC_FILES: fp = latest / fname if not fp.exists(): continue try: obj = json.loads(fp.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): continue rel = str(fp.relative_to(repo_root)) out.append((rel, obj)) return out # ============================================================================ # CLI + main # ============================================================================ def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: ap = argparse.ArgumentParser( description=( "background topology planner experiment (W1 dry-run, evidence " "pack only). DB read-only, network 0, image 0, sibling " "experiment imports 0." ), ) ap.add_argument("--output-root", type=Path, default=_REPO_ROOT / DEFAULT_OUTPUT_DIR, help="output root for run dir") ap.add_argument("--run-id", type=str, default=None, help="explicit run id (default: YYYYMMDD_HHMM_)") ap.add_argument("--project-id", type=str, default=None, help="override fixture project_id") ap.add_argument("--episode-id", type=str, default=None, help="override fixture episode_id") ap.add_argument("--include-diagnostic-runs", action="store_true", help="additive-only: also feed existing sibling experiment " "JSON artifacts as source_kind=existing_artifact rows") ap.add_argument("--acceptance-mode", choices=["sample_fixture", "override"], default=None, help="strict (sample fixture, hard fail) or override " "(advisory only). Auto-set to 'override' when " "project_id/episode_id differs from fixture.") ap.add_argument("--no-serve", action="store_true", help="reserved/no-op (W1 has no built-in webserver)") ap.add_argument("--from-run-dir", type=Path, default=None, help="W1c review-cockpit mode (plan §8-E): copy 5 raw " "files byte-identical from existing W1b run dir and " "emit enhanced index.html + readiness_report.json " "into the new run dir (no DB read, no re-extraction).") return ap.parse_args(argv) def _resolve_acceptance_mode(args: argparse.Namespace, spec: SampleFixtureSpec) -> str: if args.acceptance_mode: return args.acceptance_mode # auto-detect. overridden = False if args.project_id and args.project_id != spec.project_id: overridden = True if args.episode_id and args.episode_id != spec.episode_id: overridden = True return "override" if overridden else "sample_fixture" def main(argv: Optional[list[str]] = None) -> int: args = parse_args(argv) # W1c review-cockpit mode (plan §8-E): no DB read, no re-extraction. if args.from_run_dir is not None: source_run_dir = Path(args.from_run_dir) if not source_run_dir.exists() or not source_run_dir.is_dir(): print(f"[btp][error] --from-run-dir not found or not a dir: " f"{source_run_dir}", file=sys.stderr) return 2 run_id = args.run_id or f"{_run_id()}_w1c" out_dir = args.output_root / run_id run_meta = run_from_existing_run( source_run_dir=source_run_dir, dest_run_dir=out_dir, ) print(f"[btp][w1c] run_id={run_meta['run_id']}") print(f"[btp][w1c] out_dir={out_dir}") print(f"[btp][w1c] source_run_dir={source_run_dir}") return 0 spec = build_sample_fixture_l05_spec() # Apply CLI overrides — preserve fixture identity but allow swapping # project/episode ids for ad-hoc runs (auto-toggles acceptance_mode to # override). effective_project_id = args.project_id or spec.project_id effective_episode_id = args.episode_id or spec.episode_id # If override, build a shallow copy of spec with overridden IDs. if (effective_project_id != spec.project_id or effective_episode_id != spec.episode_id): spec = SampleFixtureSpec( fixture_id=f"{spec.fixture_id}_override", project_id=effective_project_id, episode_id=effective_episode_id, canon_id=spec.canon_id, location_short_id=spec.location_short_id, source_run_path=spec.source_run_path, source_bible_filename=spec.source_bible_filename, evidence_lexicon=spec.evidence_lexicon, ) acceptance_mode = _resolve_acceptance_mode(args, spec) ruleset = build_evidence_extraction_ruleset(spec) # ---- DB read-only ------------------------------------------------------ from app.core.database import SessionLocal source_pack = SourcePack() with SessionLocal() as session: ep_text, ep_meta = load_episode_fulltext(session, spec.episode_id) plan_text, plan_meta = load_planning_doc(session, spec.project_id) shots = load_selected_shots(session, spec.project_id, spec.episode_id) entities = load_entity_catalog(session, spec.project_id) locations = load_location_catalog(session, spec.project_id) source_pack.planning_doc_text = plan_text source_pack.episode_fulltext = ep_text source_pack.selected_shots = shots source_pack.entity_catalog = entities source_pack.location_catalog = locations if ep_meta.get("missing"): source_pack.missing_inputs.append({ "key": "episode_fulltext", "reason": "episode row 부재", "looked_at": ep_meta.get("id"), }) if plan_meta.get("missing"): source_pack.missing_inputs.append({ "key": "planning_doc_text", "reason": "project_registry row 부재", "looked_at": plan_meta.get("id"), }) # ---- Existing bible artifact (optional, single file) ------------------ bible_path = (_REPO_ROOT / spec.source_run_path / spec.source_bible_filename) if bible_path.exists(): try: bible_obj = json.loads(bible_path.read_text(encoding="utf-8")) rel = str(bible_path.relative_to(_REPO_ROOT)) source_pack.existing_artifacts.append((rel, bible_obj)) except (OSError, json.JSONDecodeError): pass # ---- Optional diagnostic input (additive-only) ------------------------ if args.include_diagnostic_runs: source_pack.diagnostic_artifacts = load_diagnostic_artifacts(_REPO_ROOT) # ---- Collect evidence -------------------------------------------------- evidence_rows = collect_all_evidence( source_pack=source_pack, ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) coverage = build_source_coverage_summary( evidence_rows=evidence_rows, spec=spec, acceptance_mode=acceptance_mode, ) args_dict = { "output_root": str(args.output_root), "run_id": args.run_id, "project_id": args.project_id, "episode_id": args.episode_id, "include_diagnostic_runs": args.include_diagnostic_runs, "acceptance_mode": acceptance_mode, "no_serve": args.no_serve, } manifest = build_input_manifest( spec=spec, source_pack=source_pack, ruleset=ruleset, args_dict=args_dict, ) run_id = args.run_id or _run_id() out_dir = args.output_root / run_id run_meta = build_run_meta( run_id=run_id, fixture_id=spec.fixture_id, args_dict=args_dict, outputs=[ "input_manifest.json", "source_evidence_pack.json", "source_evidence.tsv", "evidence_extraction_ruleset.json", "source_coverage_summary.json", "index.html", "run_meta.json", ], lexicon_hash=ruleset["lexicon_hash"], ) html = render_html( evidence_rows=evidence_rows, ruleset=ruleset, coverage=coverage, spec=spec, run_meta=run_meta, ) write_outputs( run_dir=out_dir, evidence_rows=evidence_rows, ruleset=ruleset, input_manifest=manifest, coverage_summary=coverage, html=html, run_meta=run_meta, ) # ---- Stdout summary ---------------------------------------------------- print(f"[btp] run_id={run_id}") print(f"[btp] out_dir={out_dir}") print(f"[btp] evidence_count={len(evidence_rows)}" f" / acceptance_mode={acceptance_mode}" f" / hard_pass={coverage['acceptance']['hard_pass']}") if coverage["acceptance"].get("hard_fails"): for f in coverage["acceptance"]["hard_fails"]: print(f"[btp][hard_fail] {f}") return 0 if coverage["acceptance"]["hard_pass"] else 1 if __name__ == "__main__": sys.exit(main())