"""TDD invariants for experiment_background_topology_planner (W1 dry-run). plan SOT: scripts_output/background_topology_planner_experiment/plan.md (W0b APPROVED_FOR_W1). 본 test 파일은 plan §4 의 28 invariants 와 1:1 매칭한다. 핵심 invariants: - 4-A: import / DB-write / sibling guards (no production pipeline, no image network, no sibling experiments, no DB write). - 4-B: sample fixture isolation — generic collector body 가 fixture literal 을 보지 않음. - 4-C: evidence row schema + determinism (BLOCKING 1 — evidence_id sha256, manifest hash). - 4-D: W1 scope guard — topology/unit/ledger artifact 0, ruleset 산출 1. - 4-E: diagnostic additive / no-serve / dry-run network 0. - 4-F: acceptance metric — sample_fixture strict, override advisory. - 4-G: HTML / plan.md 표기. No API calls, no DB writes, no image generation. """ from __future__ import annotations import ast import hashlib import json import re import sys from copy import deepcopy from dataclasses import asdict, is_dataclass from pathlib import Path import pytest _REPO_ROOT = Path(__file__).resolve().parents[3] _SCRIPTS = _REPO_ROOT / "backend" / "scripts" if str(_SCRIPTS) not in sys.path: sys.path.insert(0, str(_SCRIPTS)) import experiment_background_topology_planner as btp # noqa: E402 # --------------------------------------------------------------------------- # Module-level constants used by multiple test groups. # --------------------------------------------------------------------------- SCRIPT_PATH = _SCRIPTS / "experiment_background_topology_planner.py" PLAN_PATH = ( _REPO_ROOT / "scripts_output" / "background_topology_planner_experiment" / "plan.md" ) # Sample-fixture-specific literals — generic collector body must NOT contain # these. SampleFixtureSpec / EvidenceLexicon / output data are exempt. FORBIDDEN_LEAKAGE_LITERALS = [ "L05", "옥탑방", "수리영", "민숙", "안방", "pg_rooftop_villa", "sg_rooftop_interior", ] GENERIC_FUNCTION_PREFIXES = ( "collect_", "build_evidence_extraction_ruleset", "load_episode_fulltext", "load_planning_doc", "load_selected_shots", "load_entity_catalog", "load_location_catalog", ) def _module_source() -> str: return SCRIPT_PATH.read_text(encoding="utf-8") def _module_ast() -> ast.Module: return ast.parse(_module_source()) def _generic_function_nodes() -> list[ast.FunctionDef]: """All top-level function defs whose names start with a generic prefix. These must remain sample-agnostic — no fixture literal allowed anywhere in their body. """ out: list[ast.FunctionDef] = [] for node in _module_ast().body: if isinstance(node, ast.FunctionDef) and node.name.startswith( GENERIC_FUNCTION_PREFIXES, ): out.append(node) return out # --------------------------------------------------------------------------- # 4-A. Import / DB-write / Sibling-experiment guards # --------------------------------------------------------------------------- class TestStaticImportGuards: """plan §4-A 1-4: import scope + DB write 정적 검증.""" def test_script_does_not_import_production_pipeline(self): """4-A-1: `from app.` 허용 범위 = database/models.project/models.catalog.""" allowed = { "app.core.database", "app.models.project", "app.models.catalog", } tree = _module_ast() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): mod = node.module or "" if mod.startswith("app."): assert mod in allowed, ( f"forbidden production import: {mod!r}" ) elif isinstance(node, ast.Import): for alias in node.names: if alias.name.startswith("app."): assert alias.name in allowed, ( f"forbidden production import: {alias.name!r}" ) def test_script_does_not_import_image_or_network(self): """4-A-2: openai/google.genai/google.generativeai/fal/PIL/requests/httpx import 0. """ forbidden_prefixes = ( "openai", "google.genai", "google.generativeai", "fal_client", "fal.", "PIL", "requests", "httpx", ) tree = _module_ast() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): mod = node.module or "" for p in forbidden_prefixes: assert not mod.startswith(p), ( f"forbidden import-from module: {mod!r}" ) elif isinstance(node, ast.Import): for alias in node.names: for p in forbidden_prefixes: assert not alias.name.startswith(p), ( f"forbidden import module: {alias.name!r}" ) def test_script_does_not_import_sibling_experiments(self): """4-A-3: sibling experiment 모듈 import 0 (BLOCKING 3).""" sibling_modules = { "experiment_rooftop_spatial_bg", "experiment_background_place_grouping", "experiment_background_spatial_decision", "experiment_rooftop_source_grounding", } tree = _module_ast() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): mod = node.module or "" assert mod not in sibling_modules, ( f"forbidden sibling-experiment import: {mod!r}" ) elif isinstance(node, ast.Import): for alias in node.names: assert alias.name not in sibling_modules, ( f"forbidden sibling-experiment import: {alias.name!r}" ) def test_script_does_not_call_db_write(self): """4-A-4: session.add/commit/flush/delete + raw SQL write 패턴 0.""" body = _module_source() forbidden_patterns = [ "session.add(", "session.commit(", "session.flush(", "session.delete(", "INSERT ", "UPDATE ", "DELETE FROM", ] for pat in forbidden_patterns: assert pat not in body, ( f"forbidden DB-write pattern in script: {pat!r}" ) # --------------------------------------------------------------------------- # 4-B. Sample fixture isolation / generic rule body guard # --------------------------------------------------------------------------- class TestSampleFixtureIsolation: """plan §4-B 5-8: SAMPLE_FIXTURE namespacing + generic body literal 차단.""" def test_sample_fixture_constants_namespaced(self): """4-B-5: SAMPLE_FIXTURE_* + SampleFixtureSpec + EvidenceLexicon 정의 존재.""" assert hasattr(btp, "SAMPLE_FIXTURE_PROJECT_ID") assert hasattr(btp, "SAMPLE_FIXTURE_EPISODE_ID") # plan §1 references SAMPLE_FIXTURE_L05_SHORT_ID; legacy # SAMPLE_FIXTURE_SHORT_ID alias is also accepted. assert (hasattr(btp, "SAMPLE_FIXTURE_L05_SHORT_ID") or hasattr(btp, "SAMPLE_FIXTURE_SHORT_ID")) assert hasattr(btp, "SampleFixtureSpec") assert hasattr(btp, "EvidenceLexicon") # build_sample_fixture_l05_spec must be defined (BLOCKING 3 — generic # loader/collector takes spec, fixture spec ships as a single function). assert hasattr(btp, "build_sample_fixture_l05_spec") def test_scope_leakage_phrases_not_in_generic_rule_body(self): """4-B-6: generic 함수 body (AST) 안에 FORBIDDEN_LEAKAGE_LITERALS 0.""" generic_nodes = _generic_function_nodes() assert generic_nodes, ( "expected at least one generic function (collect_*, load_*, " "build_evidence_extraction_ruleset) — none found" ) for fn in generic_nodes: # Re-render the function body source and string-search for any # forbidden literal. If found, fail with the function name. fn_source = ast.unparse(fn) for lit in FORBIDDEN_LEAKAGE_LITERALS: assert lit not in fn_source, ( f"forbidden sample-fixture literal {lit!r} found in " f"generic function {fn.name!r}; move it to " f"SampleFixtureSpec/EvidenceLexicon only" ) def test_collector_body_takes_ruleset_argument(self): """4-B-7: 모든 generic collector signature 가 ruleset+lexicon param, body 에 match_terms literal 정의 0. """ import inspect collectors = [ name for name in dir(btp) if name.startswith("collect_evidence_from_") ] assert collectors, "no collect_evidence_from_* functions found" for name in collectors: fn = getattr(btp, name) sig = inspect.signature(fn) params = sig.parameters assert ("ruleset" in params) or ("rule_set" in params), ( f"{name} must accept ruleset (or rule_set) parameter" ) assert "lexicon" in params, ( f"{name} must accept lexicon parameter" ) # No collector body declares `match_terms = [...]` literal directly. for fn_node in _generic_function_nodes(): if not fn_node.name.startswith("collect_evidence_from_"): continue for sub in ast.walk(fn_node): if isinstance(sub, ast.Assign): for tgt in sub.targets: if (isinstance(tgt, ast.Name) and tgt.id == "match_terms"): # body-local term literal 정의 금지. assert not isinstance(sub.value, (ast.List, ast.Tuple)), ( f"{fn_node.name}: must not declare local " f"match_terms literal; consume from ruleset" ) def test_fixture_swap_does_not_require_collector_edit(self): """4-B-8: 합성 EvidenceLexicon 으로 fixture 갈아끼우면 collector 코드 수정 없이 다른 term 으로 evidence 추출. """ # Build a synthetic alt fixture + lexicon — different vocabulary. spec = btp.build_sample_fixture_l05_spec() alt_lexicon = btp.EvidenceLexicon( place_terms=[("alt_place_kw", "place_hint")], space_terms=[("alt_space_kw", "space_hint")], state_terms=[("alt_state_kw", "state_hint")], camera_terms=[("alt_camera_kw", "camera_hint")], ) alt_spec = btp.SampleFixtureSpec( fixture_id="alt_synth", project_id="x", episode_id="y", canon_id="z", location_short_id="LXX", source_run_path=spec.source_run_path, source_bible_filename=spec.source_bible_filename, evidence_lexicon=alt_lexicon, ) ruleset = btp.build_evidence_extraction_ruleset(alt_spec) # Sanity: synthetic terms have at least one rule each. rule_terms = { term for rule in ruleset["rules"] for term in (rule.get("match_terms") or []) } assert "alt_place_kw" in rule_terms, ( "alt fixture lexicon must drive ruleset terms without " "collector code edit" ) # Run the planning_doc collector with synthetic text containing the # alt term — must produce ≥1 evidence row. rows = btp.collect_evidence_from_planning_doc( text="some sentence with alt_place_kw inside.", ruleset=ruleset, lexicon=alt_lexicon, spec=alt_spec, ) assert any( r["evidence_type"] == "place_hint" for r in rows ), f"alt lexicon term must yield evidence rows: got {rows!r}" # --------------------------------------------------------------------------- # 4-C. Evidence row schema / determinism # --------------------------------------------------------------------------- class TestEvidenceRowSchema: """plan §4-C 9-18: evidence row 필드 + enum + deterministic id.""" def _sample_text_rows(self) -> list[dict]: spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) text = ( "옥상 위의 작은 방. 거실에 식탁이 보인다. " "카메라는 wide 로 천장을 잡는다. " "문 옆 창문으로 황혼빛이 든다." ) return btp.collect_evidence_from_planning_doc( text=text, ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) def test_evidence_row_required_fields(self): """4-C-9: 8 필수 필드.""" rows = self._sample_text_rows() assert rows, "expected ≥1 evidence row from sample planning text" required = { "evidence_id", "source_kind", "source_ref", "source_span", "quote", "evidence_type", "confidence_band", "extracted_by", } for row in rows: missing = required - set(row.keys()) assert not missing, ( f"evidence row missing required fields: {missing}; row={row}" ) def test_evidence_source_kind_enum(self): """4-C-10: source_kind enum.""" valid = { "planning_doc", "episode_fulltext", "shot_description", "scene_summary", "entity_catalog", "location_catalog", "existing_artifact", } rows = self._sample_text_rows() for r in rows: assert r["source_kind"] in valid, ( f"invalid source_kind={r['source_kind']!r}" ) def test_evidence_type_enum(self): """4-C-11: evidence_type enum.""" valid = { "place_hint", "set_hint", "space_hint", "zone_hint", "boundary_hint", "door_window_hint", "furniture_hint", "state_hint", "camera_hint", "movement_hint", "entity_layout_hint", "ambiguity_hint", } rows = self._sample_text_rows() for r in rows: assert r["evidence_type"] in valid, ( f"invalid evidence_type={r['evidence_type']!r}" ) def test_evidence_confidence_band_enum(self): """4-C-12: confidence_band enum.""" valid = {"observed", "inferred_candidate", "ambiguous"} rows = self._sample_text_rows() for r in rows: assert r["confidence_band"] in valid, ( f"invalid confidence_band={r['confidence_band']!r}" ) def test_inference_basis_required_when_not_observed(self): """4-C-13: confidence_band != observed → inference_basis + match_terms 필수 (IMPORTANT 2). """ rows = self._sample_text_rows() for r in rows: if r["confidence_band"] == "observed": continue assert r.get("inference_basis"), ( f"non-observed row missing inference_basis: {r}" ) assert r.get("match_terms"), ( f"non-observed row missing match_terms: {r}" ) def test_evidence_id_is_deterministic_sha256_prefix(self): """4-C-14: ev_<16-hex>, 두 run 결과 동일, quote 변경 시 id 변경 (BLOCKING 1). """ rows1 = self._sample_text_rows() rows2 = self._sample_text_rows() ids1 = sorted(r["evidence_id"] for r in rows1) ids2 = sorted(r["evidence_id"] for r in rows2) assert ids1 == ids2, "evidence_id must be deterministic across runs" # 형식 = ev_ + 16-hex. for eid in ids1: assert eid.startswith("ev_"), eid tail = eid[len("ev_"):] assert len(tail) == 16, eid assert all(c in "0123456789abcdef" for c in tail), eid # normalized_quote 변경 시 id 변경. sample = dict(rows1[0]) sample_quote = sample["normalized_quote"] forged = btp._make_evidence_id( source_kind=sample["source_kind"], source_ref=sample["source_ref"], source_span=sample["source_span"], evidence_type=sample["evidence_type"], normalized_quote=sample_quote + " EXTRA", ) assert forged != sample["evidence_id"], ( "changing normalized_quote must change evidence_id" ) def test_evidence_id_stable_under_extraction_order_shuffle(self): """4-C-15: collector 출력 list 셔플 후 evidence_id 집합 동일.""" rows = self._sample_text_rows() ids_before = sorted(r["evidence_id"] for r in rows) import random shuffled = list(rows) random.Random(42).shuffle(shuffled) ids_after = sorted(r["evidence_id"] for r in shuffled) assert ids_before == ids_after, ( "evidence_id set must be stable under shuffle" ) def test_evidence_id_unique_within_run(self): """4-C-16: 한 run 안에서 evidence_id 충돌 0 (dedup 강제).""" rows = self._sample_text_rows() ids = [r["evidence_id"] for r in rows] assert len(ids) == len(set(ids)), ( f"duplicate evidence_id in single run: counts={len(ids)} " f"unique={len(set(ids))}" ) def test_source_span_kind_branch_fields_required(self): """4-C-17: source_span 분기별 필수 필드.""" rows = self._sample_text_rows() for r in rows: span = r["source_span"] assert span.get("source_hash"), ( f"source_span.source_hash required: {r}" ) kind = span["kind"] if kind == "text_offset": assert span.get("char_start") is not None, r assert span.get("char_end") is not None, r elif kind == "row_pointer": assert span.get("row_id"), r assert span.get("row_field"), r elif kind == "artifact_pointer": assert span.get("artifact_path"), r assert span.get("artifact_pointer"), r else: pytest.fail(f"unknown span kind: {kind!r}") def test_input_manifest_includes_per_source_hash_and_length(self, tmp_path): """4-C-18: input_manifest.json 각 source entry 에 sha256 + length 또는 row_count + lexicon_hash mirror (BLOCKING 1). """ spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) # Build a synthetic SourcePack — bypass DB. source_pack = btp.SourcePack( planning_doc_text="a small planning doc", episode_fulltext="episode body", selected_shots=[ {"still_id": "stub", "shot_description": "wide", "scene_summary": "sum", "visible_entities_json": "[]", "scene_index": 1, "shot_index": 1, "is_selected": True, "status": "completed"}, ], entity_catalog=[ {"canon_id": "e1", "entity_type": "character", "name": "x", "description": "y", "metadata_json": "{}"}, ], location_catalog=[ {"canon_id": "L1", "entity_type": "location", "name": "l", "description": "loc desc", "metadata_json": "{}"}, ], existing_artifacts=[], diagnostic_artifacts=[], ) manifest = btp.build_input_manifest( spec=spec, source_pack=source_pack, ruleset=ruleset, args_dict={"acceptance_mode": "sample_fixture"}, ) for key in ("planning_doc", "episode_fulltext", "selected_shots", "entity_catalog", "location_catalog"): assert key in manifest["sources"], key entry = manifest["sources"][key] assert "sha256" in entry, entry assert ("length_chars" in entry) or ("row_count" in entry), entry assert manifest["lexicon_hash"] == ruleset["lexicon_hash"], ( "manifest lexicon_hash must mirror ruleset lexicon_hash" ) # --------------------------------------------------------------------------- # 4-D. W1 scope (no topology / units / decisions) # --------------------------------------------------------------------------- class TestW1ScopeGuard: """plan §4-D 19-20: future contract artifact 0 + ruleset 산출 1.""" def test_w1_does_not_emit_topology_or_units(self, tmp_path): """4-D-19: 산출 디렉토리에 topology/unit/ledger artifact 0.""" btp.write_outputs( run_dir=tmp_path, evidence_rows=[], ruleset={ "ruleset_id": "rs_test", "ruleset_version": "1.0", "lexicon_hash": btp._sha256_hex(""), "rules": [], }, input_manifest={"sources": {}, "lexicon_hash": btp._sha256_hex("")}, coverage_summary={ "evidence_count": 0, "source_kind_counts": {}, "evidence_type_counts": {}, "confidence_band_counts": {}, "leakage_guard_clean": True, "acceptance_mode": "sample_fixture", "ambiguous_rows": [], }, html="", run_meta={"run_id": "test", "plan_version": "btp_w1", "generated_at": "2026-05-24T00:00:00Z", "fixture_id": "l05_rooftop_interior", "args": {}, "outputs": [], "lexicon_hash": btp._sha256_hex("")}, ) forbidden = { "set_topology.json", "background_unit_need.json", "spatial_decision_ledger.json", "shot_spatial_intent.json", "structural_state_model.json", } for name in forbidden: assert not (tmp_path / name).exists(), ( f"W1 must not emit future contract artifact {name}" ) def test_w1_emits_evidence_extraction_ruleset(self, tmp_path): """4-D-20: evidence_extraction_ruleset.json 존재 (BLOCKING 2).""" btp.write_outputs( run_dir=tmp_path, evidence_rows=[], ruleset={ "ruleset_id": "rs_test", "ruleset_version": "1.0", "lexicon_hash": btp._sha256_hex(""), "rules": [], }, input_manifest={"sources": {}, "lexicon_hash": btp._sha256_hex("")}, coverage_summary={ "evidence_count": 0, "source_kind_counts": {}, "evidence_type_counts": {}, "confidence_band_counts": {}, "leakage_guard_clean": True, "acceptance_mode": "sample_fixture", "ambiguous_rows": [], }, html="", run_meta={"run_id": "test", "plan_version": "btp_w1", "generated_at": "2026-05-24T00:00:00Z", "fixture_id": "l05_rooftop_interior", "args": {}, "outputs": [], "lexicon_hash": btp._sha256_hex("")}, ) assert (tmp_path / "evidence_extraction_ruleset.json").exists(), ( "W1 must emit evidence_extraction_ruleset.json" ) # --------------------------------------------------------------------------- # 4-E. Diagnostic / CLI / dry-run guards # --------------------------------------------------------------------------- class TestDiagnosticAndCliGuards: """plan §4-E 21-24.""" def test_diagnostic_input_only_adds_existing_artifact_rows(self, tmp_path): """4-E-21: --include-diagnostic-runs on vs off → 차이 = existing_artifact row 만 추가, 나머지 분포 동일 (IMPORTANT 4). """ spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) # Two synthetic SourcePack instances — same base inputs, one with # diagnostic_artifacts populated. base_kwargs = dict( planning_doc_text="옥상 위 작은 방. 거실에 식탁.", episode_fulltext="문 옆 창문. 카메라는 wide.", selected_shots=[], entity_catalog=[], location_catalog=[], existing_artifacts=[], ) pack_off = btp.SourcePack(diagnostic_artifacts=[], **base_kwargs) # diagnostic artifact = list[(path,obj)] tuples (path is artificial). diag_obj = {"sub_spaces": [{"name": "거실"}]} pack_on = btp.SourcePack( diagnostic_artifacts=[("diag_run_a/sub.json", diag_obj)], **base_kwargs, ) rows_off = btp.collect_all_evidence( source_pack=pack_off, ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) rows_on = btp.collect_all_evidence( source_pack=pack_on, ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) non_diag_off = [r for r in rows_off if r["source_kind"] != "existing_artifact"] non_diag_on = [r for r in rows_on if r["source_kind"] != "existing_artifact"] assert (sorted(r["evidence_id"] for r in non_diag_off) == sorted(r["evidence_id"] for r in non_diag_on)), ( "non-diagnostic evidence_id set must be invariant under " "--include-diagnostic-runs" ) diag_only = [r for r in rows_on if r["source_kind"] == "existing_artifact"] assert diag_only, ( "expected ≥1 existing_artifact row when diagnostic input provided" ) def test_no_serve_flag_short_circuits_http_server(self): """4-E-22: --no-serve 시 http.server import/bind 0. 본 wave 는 web serve 자체 안 함 — http.server 모듈 import 자체 금지. """ body = _module_source() forbidden = ["http.server", "HTTPServer", "ThreadingHTTPServer", "SimpleHTTPRequestHandler"] for tok in forbidden: assert tok not in body, ( f"script must not reference {tok!r} — W1 is no-serve only" ) def test_dry_run_smoke_no_network(self, monkeypatch, tmp_path): """4-E-23: collector 1회 실행 → evidence pack > 0, urlopen/socket/ requests/httpx 호출 0. """ called = [] def _trap(name): def _inner(*a, **kw): called.append((name, a)) raise AssertionError(f"network call attempted via {name}") return _inner # Trap urllib + socket entry points commonly used by inadvertent net # access. The script should never call any of these in W1. import socket import urllib.request as _urlreq monkeypatch.setattr(_urlreq, "urlopen", _trap("urllib.urlopen")) monkeypatch.setattr(socket, "create_connection", _trap("socket.create_connection")) spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) # Run an in-memory collector — must not touch the network. rows = btp.collect_evidence_from_planning_doc( text="옥상에 작은 방. 거실에 창문.", ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) assert len(rows) > 0, "synthetic planning text must yield evidence" assert not called, f"unexpected network calls: {called}" def test_run_meta_contains_required_keys(self): """4-E-24: run_meta 필수 키.""" rm = btp.build_run_meta( run_id="r1", fixture_id="l05_rooftop_interior", args_dict={"x": 1}, outputs=["a.json"], lexicon_hash=btp._sha256_hex(""), ) required = {"run_id", "plan_version", "generated_at", "fixture_id", "args", "outputs", "lexicon_hash"} assert required.issubset(rm.keys()), ( f"run_meta missing keys: {required - set(rm.keys())}" ) # --------------------------------------------------------------------------- # 4-F. Acceptance metric # --------------------------------------------------------------------------- class TestAcceptanceMetric: """plan §4-F 25-26.""" def _coverage(self, mode: str, rows=None): spec = btp.build_sample_fixture_l05_spec() rows = rows or [] return btp.build_source_coverage_summary( evidence_rows=rows, spec=spec, acceptance_mode=mode, ) def test_acceptance_strict_only_for_sample_fixture_run(self): """4-F-25: sample_fixture 모드 = strict, override = advisory (hard fail 0).""" cov_strict = self._coverage("sample_fixture") cov_adv = self._coverage("override") assert cov_strict["acceptance_mode"] == "sample_fixture" assert cov_adv["acceptance_mode"] == "override" # strict 모드는 evidence_count 0 + source_kind 5종 미충족 시 hard fail. assert cov_strict["acceptance"]["hard_pass"] is False, ( "sample_fixture mode with 0 rows must NOT hard-pass" ) # override 모드는 advisory only — hard_pass 강제 X. assert "advisory_notes" in cov_adv, ( "override mode must surface advisory_notes instead of hard fail" ) def test_location_catalog_source_kind_present_for_sample_fixture(self): """4-F-26: sample fixture run 에서 location_catalog row ≥ 1. (IMPORTANT 1) """ spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) # synthetic location row. rows = btp.collect_evidence_from_location_catalog( entities=[{ "canon_id": "L05", "entity_type": "location", "name": "어떤 방", "description": "옥상 위의 작은 방", "metadata_json": json.dumps({"space_profile": {"x": 1}}, ensure_ascii=False), }], ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) assert any(r["source_kind"] == "location_catalog" for r in rows), ( "location_catalog source_kind must yield ≥1 row" ) # --------------------------------------------------------------------------- # 4-G. HTML / plan.md # --------------------------------------------------------------------------- class TestHtmlAndPlanText: """plan §4-G 27-28.""" def test_index_html_declares_generic_scope_and_w1_only(self): """4-G-27: HTML 본문에 4개 문구 명시.""" spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) coverage = btp.build_source_coverage_summary( evidence_rows=[], spec=spec, acceptance_mode="sample_fixture", ) html = btp.render_html( evidence_rows=[], ruleset=ruleset, coverage=coverage, spec=spec, run_meta={"run_id": "rtest", "plan_version": "btp_w1", "generated_at": "2026-05-24T00:00:00Z", "fixture_id": "l05_rooftop_interior", "args": {}, "outputs": [], "lexicon_hash": ruleset["lexicon_hash"]}, ) body = html.lower() assert "generic" in body, "HTML must declare generic scope" assert "sample fixture" in body, "HTML must mark sample fixture" # W1c (plan §8-E) re-labelled HTML as "review cockpit over approved # W1b snapshot"; accept either historical wording or the new one. assert ("w1 = evidence pack only" in body or "w1c = review cockpit" in body), ( "HTML must mark W1 wave scope (evidence pack only) or the W1c " "review cockpit label (plan §8-E)" ) assert "next: w2 = topology candidate generation" in body, ( "HTML must mark next wave (W2 = topology candidate generation)" ) def test_plan_md_declares_generic_scope_and_sample_fixture(self): """4-G-28: plan.md 본문에 generic / sample fixture / topology-first / 7 contract 이름 모두 명시. """ body = PLAN_PATH.read_text(encoding="utf-8") for needle in ["generic", "sample fixture", "topology-first"]: assert needle in body.lower(), f"plan.md missing {needle!r}" # 7 contracts (§1-A ~ §1-G) — class/contract name 명시. contracts = [ "PlaceContinuityGroup", "SetTopologyGraph", "StructuralStateModel", "ShotSpatialIntent", "BackgroundUnitNeed", "SpatialDecisionLedger", "EvidenceExtractionRuleSet", ] for c in contracts: assert c in body, f"plan.md missing contract name {c!r}" # --------------------------------------------------------------------------- # 4-H. W1b narrow patch invariants (Codex W1 NEEDS_REVISION fix) # --------------------------------------------------------------------------- class TestW1bPatchInvariants: """plan §8-D: inference_basis literal-free + deterministic loader order_by + _dedup_merge inference_basis preservation + evidence_rows final sort. """ def test_inference_basis_excludes_match_term_substrings(self): """4-H-29: non-observed row 의 `inference_basis` 가 그 row 의 match_terms 값 중 어느 것도 substring 으로 포함하지 않아야 함 — sample literal 이 basis 문자열에 새지 않음을 generic 하게 강제. """ spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) text = ( "옥상 위의 작은 방. 거실에 식탁이 보인다. " "카메라는 wide 로 천장을 잡는다. " "문 옆 창문으로 황혼빛이 든다." ) rows = btp.collect_evidence_from_planning_doc( text=text, ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) non_observed = [r for r in rows if r["confidence_band"] != "observed"] assert non_observed, ( "expected ≥1 non-observed row for inference_basis substring check" ) for r in non_observed: basis = r.get("inference_basis") or "" for term in r.get("match_terms") or []: if not term: continue assert term not in basis, ( f"inference_basis must not contain match_term substring; " f"term={term!r} basis={basis!r} row={r}" ) def test_artifact_collector_inference_basis_excludes_pointer_and_term(self): """4-H-30: artifact collector 가 emit 하는 inferred row 의 basis 가 artifact pointer 와 매칭 term 모두를 substring 으로 포함하지 않아야 함. """ spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) artifacts = [ ("scripts_output/fixture/sample_bible.json", { "title": "옥상 위 방", "sub_spaces": [{"name": "거실"}, {"name": "주방"}], "rooms": "옥상에 작은 방이 있다.", }), ] rows = btp.collect_evidence_from_existing_artifact( artifacts=artifacts, ruleset=ruleset, lexicon=spec.evidence_lexicon, spec=spec, ) non_observed = [r for r in rows if r["confidence_band"] != "observed"] assert non_observed, ( "expected ≥1 non-observed artifact row for substring check" ) for r in non_observed: basis = r.get("inference_basis") or "" # No match_term substring. for term in r.get("match_terms") or []: if not term: continue assert term not in basis, ( f"artifact basis leaks match_term: term={term!r} " f"basis={basis!r}" ) # No artifact_pointer substring either — pointers like # `$.sub_spaces[0].name` are structural but we keep basis fully # generic so other fixtures stay safe. ptr = (r.get("source_span") or {}).get("artifact_pointer") or "" if ptr: assert ptr not in basis, ( f"artifact basis leaks pointer: ptr={ptr!r} " f"basis={basis!r}" ) def test_db_loaders_apply_deterministic_order_by(self): """4-H-31: load_selected_shots / load_entity_catalog / load_location_catalog 가 script 본문에서 deterministic `.order_by(` 를 선언. AST 로 각 함수 body 안 호출을 찾는다. """ tree = _module_ast() loader_names = { "load_selected_shots", "load_entity_catalog", "load_location_catalog", } found: dict[str, bool] = {n: False for n in loader_names} for node in tree.body: if not isinstance(node, ast.FunctionDef): continue if node.name not in loader_names: continue for sub in ast.walk(node): if (isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute) and sub.func.attr == "order_by"): found[node.name] = True break for name, ok in found.items(): assert ok, ( f"loader {name} must call .order_by(...) for deterministic " f"row order (plan §8-D)" ) def test_dedup_merge_preserves_inference_basis_union(self): """4-H-32: 같은 5-tuple 에 서로 다른 inference_basis 가 들어오면 결과 row 의 inference_basis 가 두 basis 를 모두 substring 으로 포함 (distinct sorted `"; "` join). """ span = { "kind": "text_offset", "char_start": 0, "char_end": 5, "row_id": None, "row_field": None, "artifact_path": None, "artifact_pointer": None, "source_hash": btp._sha256_hex("body"), } eid = btp._make_evidence_id( source_kind="planning_doc", source_ref="project:x", source_span=span, evidence_type="place_hint", normalized_quote="quote", ) row_a = { "evidence_id": eid, "source_kind": "planning_doc", "source_ref": "project:x", "source_span": span, "quote": "quote", "normalized_quote": "quote", "evidence_type": "place_hint", "candidate_contract_targets": ["A"], "confidence_band": "inferred_candidate", "match_terms": ["term_a"], "inference_basis": "basis alpha", "sample_fixture_tags": [], "extracted_by": "rule_a", } row_b = dict(row_a) row_b["candidate_contract_targets"] = ["B"] row_b["match_terms"] = ["term_b"] row_b["inference_basis"] = "basis beta" row_b["extracted_by"] = "rule_b" merged = btp._dedup_merge([row_a, row_b]) assert len(merged) == 1, f"dedup must collapse same-id rows: {merged!r}" basis = merged[0].get("inference_basis") or "" assert "basis alpha" in basis, ( f"merged basis must preserve first source: {basis!r}" ) assert "basis beta" in basis, ( f"merged basis must preserve second source: {basis!r}" ) # Empty / duplicate basis 가 noise 로 새지 않음. row_c = dict(row_a) row_c["candidate_contract_targets"] = ["C"] row_c["inference_basis"] = "" row_d = dict(row_a) row_d["candidate_contract_targets"] = ["D"] row_d["inference_basis"] = "basis alpha" merged2 = btp._dedup_merge([row_a, row_c, row_d, row_b]) basis2 = merged2[0].get("inference_basis") or "" # Distinct entries only — basis alpha 가 두 번 등장하지 않음. assert basis2.count("basis alpha") == 1, ( f"merged basis must dedupe duplicates: {basis2!r}" ) # Empty basis 를 합치지 않음 (join 결과에 leading/trailing "; " noise 0). assert not basis2.startswith(";"), ( f"empty basis must not produce leading separator: {basis2!r}" ) assert not basis2.endswith(";"), ( f"empty basis must not produce trailing separator: {basis2!r}" ) def test_write_outputs_sorts_evidence_rows_by_evidence_id(self, tmp_path): """4-H-33: write_outputs 가 source_evidence_pack.json 에 evidence_id sorted 순서로 row 를 직렬화 (cross-run byte stability 안정화). """ span = { "kind": "text_offset", "char_start": 0, "char_end": 3, "row_id": None, "row_field": None, "artifact_path": None, "artifact_pointer": None, "source_hash": btp._sha256_hex("body"), } rows = [] for q in ["quote_z", "quote_a", "quote_m"]: eid = btp._make_evidence_id( source_kind="planning_doc", source_ref="project:x", source_span=span, evidence_type="place_hint", normalized_quote=q, ) rows.append({ "evidence_id": eid, "source_kind": "planning_doc", "source_ref": "project:x", "source_span": span, "quote": q, "normalized_quote": q, "evidence_type": "place_hint", "candidate_contract_targets": [], "confidence_band": "observed", "match_terms": [], "inference_basis": "", "sample_fixture_tags": [], "extracted_by": "rule_x", }) btp.write_outputs( run_dir=tmp_path, evidence_rows=rows, ruleset={"ruleset_id": "rs_t", "ruleset_version": "1.0", "lexicon_hash": btp._sha256_hex(""), "rules": []}, input_manifest={"sources": {}, "lexicon_hash": btp._sha256_hex("")}, coverage_summary={ "evidence_count": len(rows), "source_kind_counts": {}, "evidence_type_counts": {}, "confidence_band_counts": {}, "leakage_guard_clean": True, "acceptance_mode": "sample_fixture", "ambiguous_rows": [], }, html="", run_meta={"run_id": "rt", "plan_version": "btp_w1", "generated_at": "2026-05-24T00:00:00Z", "fixture_id": "f", "args": {}, "outputs": [], "lexicon_hash": btp._sha256_hex("")}, ) pack = json.loads( (tmp_path / "source_evidence_pack.json").read_text(encoding="utf-8"), ) written_ids = [r["evidence_id"] for r in pack["rows"]] assert written_ids == sorted(written_ids), ( f"source_evidence_pack rows must be sorted by evidence_id: " f"{written_ids!r}" ) # --------------------------------------------------------------------------- # 4-I. W1c HTML / readiness enhancement invariants (plan §8-E) # --------------------------------------------------------------------------- class TestW1cReviewCockpit: """plan §8-E: W1c is review/report enhancement over an approved W1b snapshot — no re-extraction. HTML reorganized to lead with W2 readiness, noisy heuristic shown, dimension-per-readiness report. """ # ---- HTML reorganization ------------------------------------------------ def _render_html_for_spec(self, rows=None, coverage_overrides=None): spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) rows = rows or [] cov = btp.build_source_coverage_summary( evidence_rows=rows, spec=spec, acceptance_mode="sample_fixture", ) if coverage_overrides: cov.update(coverage_overrides) return btp.render_html( evidence_rows=rows, ruleset=ruleset, coverage=cov, spec=spec, run_meta={"run_id": "rc", "plan_version": "btp_w1", "generated_at": "2026-05-24T00:00:00Z", "fixture_id": spec.fixture_id, "args": {}, "outputs": [], "lexicon_hash": ruleset["lexicon_hash"]}, ) def test_html_first_data_section_is_w2_readiness_summary(self): """4-I-34: HTML 본문 첫 데이터 섹션 (헤더 다음) 이 W2 readiness summary — counts 보다 위에 노출 (plan §8-E HTML 순서). """ body = self._render_html_for_spec().lower() # Two anchors: idx_ready = body.find("w2 readiness") idx_counts = body.find("source_kind 별") if idx_counts < 0: idx_counts = body.find("source_kind") assert idx_ready >= 0, "HTML must include 'w2 readiness' summary" assert idx_counts >= 0, "HTML must include source_kind counts somewhere" assert idx_ready < idx_counts, ( f"W2 readiness summary must appear before counts; " f"ready_idx={idx_ready} counts_idx={idx_counts}" ) def test_html_warns_ambiguous_zero_means_no_classification(self): """4-I-35: ambiguous=0 옆에 'ambiguity classification 미실행' 경고.""" body = self._render_html_for_spec() assert ("ambiguity classification" in body.lower() or "ambiguity 분류" in body.lower() or "ambiguity 미실행" in body.lower()), ( "HTML must warn that ambiguous=0 does NOT mean clean — W1 is " "literal-term only and ambiguity classification is not run" ) def test_html_contains_all_w1c_sections(self): """4-I-36: §0~§8 9 섹션 모두 존재 — readiness summary, flow, counts, place/set/space evidence, state evidence, camera evidence, noisy candidates, top source quotes, detailed readiness. """ body = self._render_html_for_spec().lower() required = [ "w2 readiness", # §0 summary "흐름", # §1 flow "source_kind", # §2 counts "place / set / space", # §3 place/set/space evidence "state evidence", # §4 state "camera evidence", # §5 camera "noisy", # §6 noisy candidates "top source quotes", # §7 representative "detailed", # §8 detailed readiness ] for needle in required: assert needle in body, ( f"HTML missing W1c section anchor: {needle!r}" ) # ---- Noisy heuristic ---------------------------------------------------- def test_noisy_candidate_heuristic_flags_negation_with_match_term(self): """4-I-37: quote 안에 negation marker + row.match_terms 중 하나 동시 등장 → noisy=true. """ spec = btp.build_sample_fixture_l05_spec() assert btp._is_noisy_candidate( quote="피와 시신 없이 깨끗하게 정돈된 방", match_terms=["깨끗"], negation_terms=spec.negation_terms, ), "negation '없이' + match_term '깨끗' must trigger noisy=true" def test_noisy_heuristic_does_not_flag_bare_anbang(self): """4-I-38: quote 에 '안방' 만 있을 때 (negation marker 없음) noisy=false. bare '안 ' 패턴이 안방 substring 으로 잘못 매칭되면 안 됨. """ spec = btp.build_sample_fixture_l05_spec() assert not btp._is_noisy_candidate( quote="안방 문을 여는데 갑자기 정전이 된다.", match_terms=["안방"], negation_terms=spec.negation_terms, ), "'안방' alone (no negation marker) must NOT be flagged noisy" def test_noisy_excludes_structural_rows(self): """4-I-39: match_terms 가 비어있는 row 는 noisy 평가 대상 외 (False).""" spec = btp.build_sample_fixture_l05_spec() assert not btp._is_noisy_candidate( quote="literally anything 없이 무엇이든", match_terms=[], negation_terms=spec.negation_terms, ), "structural rows (empty match_terms) must NOT be flagged noisy" def test_noisy_heuristic_flags_english_negation_with_word_boundary(self): """4-I-40: English negation regex \\bno\\b / \\bwithout\\b etc — bare substring 매칭 금지 (e.g. 'know' 의 'no' 가 trigger 하면 안 됨). """ spec = btp.build_sample_fixture_l05_spec() # Positive: 'no blood' triggers. assert btp._is_noisy_candidate( quote="The room has no blood marks left.", match_terms=["blood"], negation_terms=spec.negation_terms, ), "\\bno\\b + match_term must trigger" # Negative: 'know' should NOT trigger. assert not btp._is_noisy_candidate( quote="We know this is a clean room.", match_terms=["room"], negation_terms=spec.negation_terms, ), "'know' substring containing 'no' must NOT trigger word-boundary" # ---- Readiness dimensions ---------------------------------------------- def test_w2_readiness_thresholds_loaded_from_spec(self): """4-I-41: readiness_thresholds 가 spec dataclass field 로만 정의되고, dimension key 가 명시 (topology_candidate / state_model / camera_intent / overall 등). script body 에 hardcoded threshold 없음. """ spec = btp.build_sample_fixture_l05_spec() assert hasattr(spec, "readiness_thresholds"), ( "SampleFixtureSpec must declare readiness_thresholds field" ) thresholds = spec.readiness_thresholds assert isinstance(thresholds, dict), thresholds for dim in ("topology_candidate", "state_model", "camera_intent"): assert dim in thresholds, ( f"readiness_thresholds missing dimension {dim!r}: " f"{thresholds!r}" ) def test_w2_readiness_dimensions_report_camera_weak_when_sparse(self): """4-I-42: synthetic rows (camera 1 term, movement 0) → camera_intent dimension status weak/no. """ spec = btp.build_sample_fixture_l05_spec() # Build a synthetic row set: rich place/space/state, sparse camera. def _row(etype, term, src="planning_doc", band="observed"): span = {"kind": "text_offset", "char_start": 0, "char_end": len(term), "row_id": None, "row_field": None, "artifact_path": None, "artifact_pointer": None, "source_hash": btp._sha256_hex("body")} return { "evidence_id": btp._make_evidence_id( source_kind=src, source_ref="x", source_span=span, evidence_type=etype, normalized_quote=term, ), "source_kind": src, "source_ref": "x", "source_span": span, "quote": term, "normalized_quote": term, "evidence_type": etype, "candidate_contract_targets": [], "confidence_band": band, "match_terms": [term] if band != "observed" else [term], "inference_basis": "" if band == "observed" else "rule_x", "sample_fixture_tags": [], "extracted_by": "rule_x", } rows = [] for term in ["place_a", "place_b"]: rows.append(_row("place_hint", term)) for term in ["space_a", "space_b", "space_c", "space_d"]: rows.append(_row("space_hint", term)) rows.append(_row("boundary_hint", "boundary_a")) rows.append(_row("door_window_hint", "door_a")) for term in ["state_a", "state_b", "state_c"]: rows.append(_row("state_hint", term)) rows.append(_row("camera_hint", "close-up")) # only 1 camera term # No movement_hint at all. report = btp._evaluate_readiness(evidence_rows=rows, spec=spec) assert "camera_intent" in report, report cam_status = report["camera_intent"]["status"] assert cam_status in {"weak", "no"}, ( f"camera_intent must be weak/no when sparse: got {cam_status!r} " f"report={report['camera_intent']!r}" ) # Overall must downgrade to conditional/no (worst dimension). overall = report.get("overall", {}).get("status") assert overall in {"conditional", "no", "weak"}, ( f"overall readiness must downgrade when any dim weak/no: " f"got {overall!r}" ) # ---- --from-run-dir review mode (raw byte-identical) ------------------- def _write_synthetic_source_run(self, dst: Path): """Materialize a minimal W1b-style source run dir (5 raw files + index.html + run_meta.json) for from-run-dir tests. """ spec = btp.build_sample_fixture_l05_spec() ruleset = btp.build_evidence_extraction_ruleset(spec) manifest = btp.build_input_manifest( spec=spec, source_pack=btp.SourcePack( planning_doc_text="옥상 위 거실에 식탁.", episode_fulltext="문 옆 창문.", ), ruleset=ruleset, args_dict={"acceptance_mode": "sample_fixture"}, ) cov = btp.build_source_coverage_summary( evidence_rows=[], spec=spec, acceptance_mode="sample_fixture", ) run_meta = btp.build_run_meta( run_id="src_run_0", fixture_id=spec.fixture_id, args_dict={"acceptance_mode": "sample_fixture"}, 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"], ) btp.write_outputs( run_dir=dst, evidence_rows=[], ruleset=ruleset, input_manifest=manifest, coverage_summary=cov, html="w1b", run_meta=run_meta, ) def test_w1c_copies_raw_artifacts_byte_identical_from_source_run( self, tmp_path): """4-I-43: --from-run-dir 가 src 의 5 raw files 를 byte-identical copy. """ src = tmp_path / "src_w1b" dst = tmp_path / "dst_w1c" self._write_synthetic_source_run(src) btp.run_from_existing_run(source_run_dir=src, dest_run_dir=dst) raw_files = [ "source_evidence_pack.json", "source_evidence.tsv", "evidence_extraction_ruleset.json", "input_manifest.json", "source_coverage_summary.json", ] for fname in raw_files: src_sha = hashlib.sha256((src / fname).read_bytes()).hexdigest() dst_sha = hashlib.sha256((dst / fname).read_bytes()).hexdigest() assert src_sha == dst_sha, ( f"W1c must byte-identical-copy {fname}: " f"src={src_sha[:16]} dst={dst_sha[:16]}" ) def test_w1c_does_not_modify_source_run_raw_files(self, tmp_path): """4-I-44: --from-run-dir 후 source run 의 5 raw 파일 sha 변경 0.""" src = tmp_path / "src_w1b" dst = tmp_path / "dst_w1c" self._write_synthetic_source_run(src) raw_files = [ "source_evidence_pack.json", "source_evidence.tsv", "evidence_extraction_ruleset.json", "input_manifest.json", "source_coverage_summary.json", ] before = {f: hashlib.sha256((src / f).read_bytes()).hexdigest() for f in raw_files} btp.run_from_existing_run(source_run_dir=src, dest_run_dir=dst) after = {f: hashlib.sha256((src / f).read_bytes()).hexdigest() for f in raw_files} for f in raw_files: assert before[f] == after[f], ( f"W1c must NOT touch source-run {f}: before={before[f][:16]} " f"after={after[f][:16]}" ) def test_w1c_emits_readiness_report_json(self, tmp_path): """4-I-45: W1c run dir 에 readiness_report.json 신규 산출 — dimension 별 status + threshold + observed values 포함. """ src = tmp_path / "src_w1b" dst = tmp_path / "dst_w1c" self._write_synthetic_source_run(src) btp.run_from_existing_run(source_run_dir=src, dest_run_dir=dst) rp = dst / "readiness_report.json" assert rp.exists(), "W1c must emit readiness_report.json" report = json.loads(rp.read_text(encoding="utf-8")) for dim in ("topology_candidate", "state_model", "camera_intent", "overall"): assert dim in report, ( f"readiness_report.json missing dimension {dim!r}" ) assert "status" in report[dim], ( f"dimension {dim} missing status field" ) def test_w1c_run_meta_marks_review_mode(self, tmp_path): """4-I-46: W1c run_meta.json 가 plan_version='btp_w1c' 또는 review mode 마커 + source_run_dir reference 포함. """ src = tmp_path / "src_w1b" dst = tmp_path / "dst_w1c" self._write_synthetic_source_run(src) btp.run_from_existing_run(source_run_dir=src, dest_run_dir=dst) rm = json.loads((dst / "run_meta.json").read_text(encoding="utf-8")) plan_ver = rm.get("plan_version") or "" assert "w1c" in plan_ver.lower(), ( f"W1c run_meta plan_version must mark W1c: {plan_ver!r}" ) assert rm.get("source_run_dir") or rm.get("source_run"), ( f"W1c run_meta must reference source_run_dir: keys={list(rm)!r}" ) # --------------------------------------------------------------------------- # 4-J. W1d narrow patch invariants (Codex W1c NEEDS_REVISION fix, plan §8-F) # --------------------------------------------------------------------------- class TestW1dPatchInvariants: """plan §8-F: state_model 의 noisy>0 시 conditional 다운그레이드 + boundary/door_window metric key 의 의미 명확화 (row count → `*_rows`). """ def _make_row(self, etype: str, term: str, *, quote: Optional[str] = None, band: str = "observed", source_kind: str = "planning_doc") -> dict: q = quote if quote is not None else term span = { "kind": "text_offset", "char_start": 0, "char_end": len(term), "row_id": None, "row_field": None, "artifact_path": None, "artifact_pointer": None, "source_hash": btp._sha256_hex("body"), } return { "evidence_id": btp._make_evidence_id( source_kind=source_kind, source_ref="x", source_span=span, evidence_type=etype, normalized_quote=q, ), "source_kind": source_kind, "source_ref": "x", "source_span": span, "quote": q, "normalized_quote": q, "evidence_type": etype, "candidate_contract_targets": [], "confidence_band": band, "match_terms": [term], "inference_basis": "" if band == "observed" else "rule_x", "sample_fixture_tags": [], "extracted_by": "rule_x", } def _baseline_topology_rows(self) -> list[dict]: """Rows that on their own satisfy topology_candidate fully.""" rows: list[dict] = [] for t in ["p1", "p2"]: rows.append(self._make_row("place_hint", t)) for t in ["s1", "s2", "s3", "s4"]: rows.append(self._make_row("space_hint", t)) rows.append(self._make_row("boundary_hint", "b1")) rows.append(self._make_row("door_window_hint", "dw1")) # camera coverage so the test isolates state_model. for t in ["wide", "close-up"]: rows.append(self._make_row("camera_hint", t)) rows.append(self._make_row("movement_hint", "pan")) return rows def test_state_model_conditional_when_noisy_rows_positive(self): """4-J-47: state_model 이 coverage threshold 충족 + noisy_rows > 0 → status='conditional' (yes 가 아니어야 함). missing/note 에 noisy row 표시. """ spec = btp.build_sample_fixture_l05_spec() rows = self._baseline_topology_rows() # state coverage: 3 distinct terms — meets threshold. rows.append(self._make_row("state_hint", "정돈")) rows.append(self._make_row("state_hint", "어두운")) # noisy state: '깨끗' inside a quote with negation '없이'. rows.append(self._make_row( "state_hint", "깨끗", quote="피와 시신 없이 깨끗하게 정돈된 방", )) report = btp._evaluate_readiness(evidence_rows=rows, spec=spec) st = report["state_model"] assert st["state_total_rows"] >= 3, st assert st["state_noisy_rows"] >= 1, st assert st["status"] == "conditional", ( f"state_model must downgrade to conditional when noisy_rows>0; " f"got status={st['status']!r}, missing={st.get('missing')!r}" ) # Missing/note must surface the noisy row count for the reviewer. joined = " ".join(st.get("missing") or []) assert any( kw in joined for kw in ("noisy", "review", "candidates") ), f"state_model conditional must explain noisy review: {joined!r}" def test_state_model_yes_only_when_noisy_zero_and_coverage_met(self): """4-J-48: state_model='yes' 는 coverage 통과 + noisy=0 일 때만.""" spec = btp.build_sample_fixture_l05_spec() rows = self._baseline_topology_rows() for t in ["정돈", "어두운", "황혼"]: rows.append(self._make_row("state_hint", t)) report = btp._evaluate_readiness(evidence_rows=rows, spec=spec) st = report["state_model"] assert st["state_noisy_rows"] == 0, st assert st["status"] == "yes", ( f"state_model must be 'yes' when coverage met AND noisy=0; " f"got {st['status']!r}" ) def test_topology_candidate_metric_keys_use_rows_for_boundary_and_door_window( self): """4-J-49: topology_candidate metric key 가 boundary/door_window 에서 row-count 의미임을 key 이름으로 명시 — `*_rows` (Codex W1c IMPORTANT 2). place/space 는 distinct term 의미를 유지. """ spec = btp.build_sample_fixture_l05_spec() rows = self._baseline_topology_rows() report = btp._evaluate_readiness(evidence_rows=rows, spec=spec) topo = report["topology_candidate"] observed = topo["observed"] threshold = topo["threshold"] # Old (misleading) keys must be gone, new keys must be present. for old_key in ("boundary_terms", "door_window_terms"): assert old_key not in observed, ( f"topology_candidate.observed must NOT use misleading " f"row-count-as-terms key {old_key!r}" ) assert old_key not in threshold, ( f"topology_candidate.threshold must NOT use misleading key " f"{old_key!r}" ) for new_key in ("boundary_rows", "door_window_rows"): assert new_key in observed, ( f"topology_candidate.observed must include row-count key " f"{new_key!r}" ) assert new_key in threshold, ( f"topology_candidate.threshold must include row-count key " f"{new_key!r}" ) # place/space distinct-term keys preserved. for keep_key in ("place_distinct_terms", "space_distinct_terms"): assert keep_key in observed, observed assert keep_key in threshold, threshold