"""TDD invariants for experiment_background_semantic_extractor (W1 dry-run).

plan SOT: scripts_output/background_semantic_extractor_experiment/plan.md
(W0g APPROVED_FOR_W1). 23 tests in 5 groups (6-A static guards / 6-B source
bundle integrity / 6-C schema shape / 6-D checker / 6-E prompt+CLI+output).

★★★ standing rules (plan §0):
  - No `re` import.
  - No lexicon/term/boundary/particle constants.
  - No classifier function definitions.
  - No checker function content scan of brief text.
  - selected_shots scoped by JSON exact equality on visible_entities_json
    (W0b BLOCKING 2).
  - Stage A → Stage B leakage = structural input contract only (W0g
    BLOCKING). No substring/content comparison.
  - All sample fixture data isolated in SAMPLE_FIXTURE_* / SampleFixtureSpec.

This test module itself MAY contain pattern words (regex, substring,
boundary, etc.) in test names and docstrings — those are documentation,
not engine code.
"""
from __future__ import annotations

import ast
import inspect
import json
import socket
import subprocess
import sys
import urllib.request as _urlreq
from copy import deepcopy
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_semantic_extractor as bsx  # noqa: E402


SCRIPT_PATH = _SCRIPTS / "experiment_background_semantic_extractor.py"

FORBIDDEN_LEAKAGE_LITERALS = [
    "L05", "옥탑방", "수리영", "민숙", "안방",
]


def _module_source() -> str:
    return SCRIPT_PATH.read_text(encoding="utf-8")


def _module_ast() -> ast.Module:
    return ast.parse(_module_source())


def _function_def_by_name(name: str) -> ast.FunctionDef | None:
    for node in _module_ast().body:
        if isinstance(node, ast.FunctionDef) and node.name == name:
            return node
    return None


# ---------------------------------------------------------------------------
# 6-A. Static pattern-recognition ban guards (5)
# ---------------------------------------------------------------------------
class TestPatternRecognitionBanGuards:
    """plan §6-A 1-5 + §6-A-6 unit pattern-word regression."""

    def test_script_has_no_re_import(self):
        """6-A-1: AST Import/ImportFrom 검사. `re` import 0."""
        tree = _module_ast()
        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom):
                mod = node.module or ""
                assert mod != "re" and not mod.startswith("re."), (
                    f"forbidden from-import: {mod!r}"
                )
            elif isinstance(node, ast.Import):
                for alias in node.names:
                    assert alias.name != "re" and not alias.name.startswith("re."), (
                        f"forbidden import: {alias.name!r}"
                    )

    def test_script_has_no_term_lexicon_or_boundary_constants(self):
        """6-A-2: module-level assignment name 검사."""
        forbidden_names = {
            "LEXICON", "TERMS", "NEGATION", "PARTICLES", "BOUNDARY",
            "HANGUL", "MIN_TERM_LENGTH", "WORD_BOUNDARY",
            "FORBIDDEN_WORDS", "DENYLIST",
        }
        tree = _module_ast()
        for node in tree.body:
            if isinstance(node, ast.Assign):
                for tgt in node.targets:
                    if isinstance(tgt, ast.Name) and tgt.id in forbidden_names:
                        pytest.fail(
                            f"forbidden module-level constant name: {tgt.id}"
                        )

    def test_script_has_no_classifier_function_definition(self):
        """6-A-4 (plan #4): function name prefix 검사."""
        forbidden_prefixes = (
            "classify_", "match_term_", "iter_term_",
            "is_valid_match_", "scan_forbidden_",
        )
        tree = _module_ast()
        for node in tree.body:
            if isinstance(node, ast.FunctionDef):
                assert not node.name.startswith(forbidden_prefixes), (
                    f"forbidden classifier function name: {node.name}"
                )

    def test_checker_function_ast_has_no_brief_content_scan(self):
        """6-A-5: run_validation 함수 AST 안에 brief item field string
        content 를 in/find/contains/lower/loop over deny_list 로 검사하는
        구조 부재. boolean check (`if not item['reasoning_basis']:`) 같은
        falsiness 확인은 허용.
        """
        run_v = _function_def_by_name("run_validation")
        assert run_v is not None, "run_validation function not found"
        for sub in ast.walk(run_v):
            # `for x in deny_list` style — Name id ending with `_list` /
            # `denylist` is suspicious.
            if isinstance(sub, ast.For):
                if isinstance(sub.iter, ast.Name) and sub.iter.id.lower() in {
                    "deny_list", "denylist", "forbidden_words",
                    "blacklist", "bad_words",
                }:
                    pytest.fail(
                        f"checker iterates over deny_list-like name: {sub.iter.id}"
                    )
            # `.find(` / `.contains(` / `.lower()` invocations applied to a
            # brief content string field.
            if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute):
                if sub.func.attr in {"find", "contains", "lower"}:
                    # Only fail if the value being called is a known brief
                    # content field name like reasoning_basis or description.
                    parent_src = ast.unparse(sub.func.value)
                    suspicious_targets = (
                        "reasoning_basis", "description", "label",
                        "one_sentence", "base_plate_prompt_brief",
                    )
                    if any(t in parent_src for t in suspicious_targets):
                        pytest.fail(
                            f"checker scans brief content via .{sub.func.attr}: "
                            f"{ast.unparse(sub)}"
                        )

    def test_checker_unit_passes_reasoning_text_with_pattern_words(self):
        """6-A-6: regression — checker does NOT scan brief text for "regex /
        boundary / literal / substring / 조사" etc. brief item containing
        those words inside description should still pass schema/provenance.
        """
        spec = bsx.build_sample_fixture_l05_spec()
        # Minimal synthetic bundle with one source whose text contains the
        # tricky words verbatim.
        source_text = (
            "literal lexicon term matched via regex boundary 조사 substring"
        )
        bundle = bsx.SourceBundle(
            bundle_id="bsx_test", fixture_id=spec.fixture_id,
            project_id=spec.project_id, episode_id=spec.episode_id,
            location_short_id=spec.location_short_id,
            sources=[
                bsx.SourceItem(
                    source_ref="planning_doc:test",
                    kind="planning_doc", text=source_text,
                    sha256=bsx._sha256_hex(source_text),
                    char_count=len(source_text),
                ),
            ],
        )
        wschema = bsx.build_world_brief_schema(bundle)
        mschema = bsx.build_minimal_spatial_brief_schema(bundle)
        evidence_ref = {
            "source_ref": "planning_doc:test",
            "quote": "literal lexicon term matched via regex boundary 조사 substring",
            "confidence": "plausible",
        }
        world_brief = {
            "schema_version": bsx.STAGE_A_SCHEMA_VERSION,
            "bundle_id": bundle.bundle_id,
            "world_background_brief": {
                "era_and_time_period": "literal regex boundary 조사 era",
                "geographic_cultural_grounding": "literal substring grounding",
                "technology_and_material_baseline": "boundary tech",
                "social_economic_visual_tone": "substring tone",
                "genre_mood_constraints": "regex genre",
                "background_relevant_do_not_assume": ["literal boundary item"],
                "irrelevant_or_do_not_pass_down": ["regex irrelevant"],
                "evidence_refs": [evidence_ref],
            },
            "confidence_band": "plausible",
        }
        minimal_brief = {
            "schema_version": bsx.STAGE_B_SCHEMA_VERSION,
            "bundle_id": bundle.bundle_id,
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": ["substring hint"],
            "place_identity": {
                "label": "regex boundary place",
                "one_sentence": "literal substring 조사 sentence",
                "evidence_refs": [evidence_ref],
            },
            "continuity_groups": [{
                "group_id": "cg_one",
                "label": "regex group",
                "what_must_stay_consistent": ["literal item"],
                "allowed_variations": ["substring item"],
                "evidence_refs": [evidence_ref],
            }],
            "essential_spatial_relations": [{
                "relation_id": "rel_one",
                "description": "literal boundary description",
                "why_it_matters_for_generation": "regex reason",
                "evidence_refs": [evidence_ref],
            }],
            "visual_anchors": [{
                "anchor_id": "anc_one",
                "description": "substring anchor",
                "role": "orientation",
                "base_plate_role": "identity_anchor",
                "evidence_refs": [evidence_ref],
            }],
            "state_variations": [{
                "state_id": "st_one",
                "description": "조사 state",
                "changes_only": ["regex change"],
                "must_not_change": ["literal invariant"],
                "evidence_refs": [evidence_ref],
            }],
            "generation_notes": {
                "base_plate_prompt_brief": "literal substring base plate",
                "shot_background_prompt_rules": ["regex rule"],
                "avoid_over_specification": ["boundary avoid"],
                "unknowns_to_keep_loose": ["조사 unknown"],
            },
            "confidence_band": "plausible",
        }
        stage_b_input = {
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": ["substring hint"],
        }
        report = bsx.run_validation(
            world_brief=world_brief, minimal_brief=minimal_brief,
            stage_b_input=stage_b_input, bundle=bundle, spec=spec,
            world_schema=wschema, minimal_schema=mschema,
        )
        assert report["passed"], (
            f"checker must pass when content has pattern-words: "
            f"failed={report['failed_checks']!r}"
        )


# ---------------------------------------------------------------------------
# 6-B. SourceBundle integrity (3) — JSON exact equality + sha256/char_count
# ---------------------------------------------------------------------------
class TestSourceBundleIntegrity:
    def test_source_bundle_selected_shots_scoped_via_json_exact_equality(self):
        """6-B-6: loader 본문에 substring matching pattern 없음.
        json.loads 사용 + dict short_id == location_short_id exact equality.
        """
        fn = _function_def_by_name("load_selected_shots_for_location")
        assert fn is not None, "loader not found"
        body_src = ast.unparse(fn)
        # No substring filter patterns:
        forbidden_patterns = [
            "if loc in ve", "in visible_entities_json", "ve.find(",
            "ve.lower(", "loc_short_id in ve",
        ]
        for pat in forbidden_patterns:
            assert pat not in body_src, (
                f"loader has forbidden substring matching pattern: {pat}"
            )
        # Must call json.loads on visible_entities_json.
        assert "json.loads" in body_src, (
            "loader must json.loads(visible_entities_json) — exact JSON parsing"
        )
        # Must compare short_id with ==
        assert "==" in body_src and "short_id" in body_src, (
            "loader must exact-equality compare short_id"
        )

    def test_source_bundle_each_source_has_ref_text_sha256_count(self):
        """6-B-7: 모든 source 4 필수 필드."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="plan body", episode_text="ep body",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        assert bundle.sources, "expected at least 2 sources (planning+episode)"
        for src in bundle.sources:
            assert src.source_ref, src
            assert isinstance(src.text, str), src
            assert isinstance(src.sha256, str) and len(src.sha256) == 64, src
            assert isinstance(src.char_count, int) and src.char_count >= 0, src

    def test_source_bundle_char_budget_estimate_present(self):
        """6-B-9: bundle 에 char_budget_estimate (int) 존재."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="abc", episode_text="defgh",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        assert isinstance(bundle.char_budget_estimate, int)
        assert bundle.char_budget_estimate >= len("abc") + len("defgh")


# ---------------------------------------------------------------------------
# 6-C. Schema shape (Stage A + Stage B) (7)
# ---------------------------------------------------------------------------
class TestSchemaShape:
    def _bundle(self):
        spec = bsx.build_sample_fixture_l05_spec()
        return spec, bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )

    def test_world_brief_schema_has_required_top_level_keys(self):
        """6-C-9: Stage A 4 키."""
        _, bundle = self._bundle()
        s = bsx.build_world_brief_schema(bundle)
        for k in ["schema_version", "bundle_id",
                  "world_background_brief", "confidence_band"]:
            assert k in s.get("required", []), k

    def test_world_brief_required_subfields(self):
        """6-C-10: world_background_brief 8 subfield."""
        _, bundle = self._bundle()
        s = bsx.build_world_brief_schema(bundle)
        wbb_req = s["world_background_brief"]["required"]
        for k in [
            "era_and_time_period", "geographic_cultural_grounding",
            "technology_and_material_baseline",
            "social_economic_visual_tone", "genre_mood_constraints",
            "background_relevant_do_not_assume",
            "irrelevant_or_do_not_pass_down", "evidence_refs",
        ]:
            assert k in wbb_req, k

    def test_minimal_brief_schema_has_required_top_level_keys(self):
        """6-C-11: Stage B 11 키."""
        _, bundle = self._bundle()
        s = bsx.build_minimal_spatial_brief_schema(bundle)
        for k in [
            "schema_version", "bundle_id", "world_brief_ref",
            "world_hints_for_background", "place_identity",
            "continuity_groups", "essential_spatial_relations",
            "visual_anchors", "state_variations",
            "generation_notes", "confidence_band",
        ]:
            assert k in s.get("required", []), k

    def test_minimal_brief_does_not_embed_full_world_context(self):
        """6-C-12: Stage B schema forbids `world_context_brief` /
        `world_background_brief` keys."""
        _, bundle = self._bundle()
        s = bsx.build_minimal_spatial_brief_schema(bundle)
        forbidden = s.get("forbidden_top_level_or_item_fields", [])
        assert "world_context_brief" in forbidden
        assert "world_background_brief" in forbidden

    def test_minimal_brief_evidence_ref_required_fields(self):
        """6-C-13: evidence_ref source_ref/quote/confidence 필수,
        char_start/end optional."""
        _, bundle = self._bundle()
        s = bsx.build_minimal_spatial_brief_schema(bundle)
        pi = s["place_identity"]
        er_schema = pi["properties"]["evidence_refs"]["items"]
        for k in ["source_ref", "quote", "confidence"]:
            assert k in er_schema["required"], k
        props = er_schema["properties"]
        assert "char_start" in props
        assert "char_end" in props
        # Optional: not in required.
        assert "char_start" not in er_schema["required"]
        assert "char_end" not in er_schema["required"]

    def test_schemas_have_no_count_json_key(self):
        """6-C-14 (W0f IMPORTANT 2): output JSON key 'count' 0. quote 본문
        에 숫자/표현은 허용 (checker 가 quote text grep 안 함).
        """
        _, bundle = self._bundle()
        for s in (bsx.build_world_brief_schema(bundle),
                   bsx.build_minimal_spatial_brief_schema(bundle)):
            # Required list 어디에도 'count' 없음
            assert "count" not in s.get("required", [])
            # forbidden list 에 count 포함
            assert "count" in s.get("forbidden_top_level_or_item_fields", []), (
                f"schema must forbid 'count' JSON key: {s}"
            )

    def test_minimal_brief_visual_anchor_uses_base_plate_role_not_must_appear(self):
        """6-C-15 (W0f IMPORTANT 3): visual_anchors item required
        base_plate_role (enum 3), no must_appear_in_base_plate field.
        """
        _, bundle = self._bundle()
        s = bsx.build_minimal_spatial_brief_schema(bundle)
        anchor_schema = s["visual_anchors"]["items"]
        assert "base_plate_role" in anchor_schema["required"]
        assert "must_appear_in_base_plate" not in anchor_schema["required"]
        assert "must_appear_in_base_plate" not in anchor_schema.get(
            "properties", {}
        )
        # Enum present.
        bpr = anchor_schema["properties"]["base_plate_role"]
        assert bpr.get("enum") == list(bsx.BASE_PLATE_ROLES)


# ---------------------------------------------------------------------------
# 6-D. Deterministic checker (provenance + shape only + cap + input contract)
# (6)
# ---------------------------------------------------------------------------
def _minimal_synthetic_bundle():
    spec = bsx.build_sample_fixture_l05_spec()
    text = "the planning text mentions a coastal village at present day."
    bundle = bsx.SourceBundle(
        bundle_id="bsx_test", fixture_id=spec.fixture_id,
        project_id=spec.project_id, episode_id=spec.episode_id,
        location_short_id=spec.location_short_id,
        sources=[
            bsx.SourceItem(
                source_ref="planning_doc:test",
                kind="planning_doc", text=text,
                sha256=bsx._sha256_hex(text), char_count=len(text),
            ),
        ],
    )
    return spec, bundle


def _basic_valid_briefs(spec, bundle):
    quote = "coastal village at present day"
    er = {"source_ref": "planning_doc:test", "quote": quote,
          "confidence": "plausible"}
    world = {
        "schema_version": bsx.STAGE_A_SCHEMA_VERSION,
        "bundle_id": bundle.bundle_id,
        "world_background_brief": {
            "era_and_time_period": "present-day",
            "geographic_cultural_grounding": "coastal village",
            "technology_and_material_baseline": "contemporary",
            "social_economic_visual_tone": "working-class",
            "genre_mood_constraints": "occult thriller",
            "background_relevant_do_not_assume": ["no luxury"],
            "irrelevant_or_do_not_pass_down": ["plot lore"],
            "evidence_refs": [er],
        },
        "confidence_band": "plausible",
    }
    minimal = {
        "schema_version": bsx.STAGE_B_SCHEMA_VERSION,
        "bundle_id": bundle.bundle_id,
        "world_brief_ref": "world_background_brief",
        "world_hints_for_background": ["coastal weathered"],
        "place_identity": {
            "label": "rooftop interior",
            "one_sentence": "small rooftop dwelling at coastal village.",
            "evidence_refs": [er],
        },
        "continuity_groups": [],
        "essential_spatial_relations": [],
        "visual_anchors": [],
        "state_variations": [],
        "generation_notes": {
            "base_plate_prompt_brief": "rooftop, weathered, coastal.",
            "shot_background_prompt_rules": [],
            "avoid_over_specification": [],
            "unknowns_to_keep_loose": [],
        },
        "confidence_band": "plausible",
    }
    stage_b_input = {
        "world_brief_ref": "world_background_brief",
        "world_hints_for_background": ["coastal weathered"],
    }
    return world, minimal, stage_b_input


class TestDeterministicChecker:
    def test_checker_validates_quote_exact_containment_in_source_text(self):
        """6-D-16: synthetic quote not in source -> fail."""
        spec, bundle = _minimal_synthetic_bundle()
        world, minimal, sbi = _basic_valid_briefs(spec, bundle)
        # Corrupt one quote.
        world["world_background_brief"]["evidence_refs"][0]["quote"] = (
            "this quote is not in the source text"
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        rep = bsx.run_validation(
            world_brief=world, minimal_brief=minimal, stage_b_input=sbi,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert not rep["passed"]
        assert any("quote_not_in_source" in f for f in rep["failed_checks"])

    def test_checker_augments_quote_occurrence_positions(self):
        """6-D-17: augmented_occurrences contains char_start/end."""
        spec, bundle = _minimal_synthetic_bundle()
        world, minimal, sbi = _basic_valid_briefs(spec, bundle)
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        rep = bsx.run_validation(
            world_brief=world, minimal_brief=minimal, stage_b_input=sbi,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert rep.get("augmented_occurrences"), rep
        for occ in rep["augmented_occurrences"]:
            assert "char_start" in occ and "char_end" in occ

    def test_checker_rejects_topology_or_count_or_embedded_world(self):
        """6-D-18: topology/count/embedded world keys -> fail."""
        spec, bundle = _minimal_synthetic_bundle()
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        # Topology field in Stage B
        world, minimal, sbi = _basic_valid_briefs(spec, bundle)
        bad_topo = deepcopy(minimal)
        bad_topo["parent_place_group_id"] = "pg_x"
        rep = bsx.run_validation(
            world_brief=world, minimal_brief=bad_topo, stage_b_input=sbi,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert not rep["passed"]
        assert any("forbidden_key" in f and "parent_place_group_id" in f
                   for f in rep["failed_checks"])
        # count JSON key
        bad_count = deepcopy(minimal)
        bad_count["count"] = 3
        rep2 = bsx.run_validation(
            world_brief=world, minimal_brief=bad_count, stage_b_input=sbi,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert any("forbidden_key" in f and "count" in f
                   for f in rep2["failed_checks"])
        # embedded world dict
        bad_world = deepcopy(minimal)
        bad_world["world_background_brief"] = {"foo": "bar"}
        rep3 = bsx.run_validation(
            world_brief=world, minimal_brief=bad_world, stage_b_input=sbi,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert any("forbidden_key" in f and "world_background_brief" in f
                   for f in rep3["failed_checks"])

    def test_checker_validates_two_schemas_independently(self):
        """6-D-19: a fail in Stage A does not block Stage B checks."""
        spec, bundle = _minimal_synthetic_bundle()
        world, minimal, sbi = _basic_valid_briefs(spec, bundle)
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        # Break Stage A required field but keep Stage B valid.
        broken_world = deepcopy(world)
        del broken_world["world_background_brief"]
        rep = bsx.run_validation(
            world_brief=broken_world, minimal_brief=minimal, stage_b_input=sbi,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert not rep["passed"]
        # Stage A missing
        assert any(f.startswith("stage_a:") for f in rep["failed_checks"])
        # Augmentation still happened for Stage B's evidence refs.
        assert rep["augmented_occurrences"], (
            "Stage B evidence refs should still be augmented despite Stage A "
            "failure"
        )

    def test_checker_enforces_minimality_caps(self):
        """6-D-20: cap boundary PASS / over-cap FAIL."""
        spec, bundle = _minimal_synthetic_bundle()
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        world, minimal, sbi = _basic_valid_briefs(spec, bundle)
        # at-cap world hint (exactly 120 chars) -> PASS
        hint_at_cap = "x" * bsx.STAGE_B_WORLD_HINTS_ITEM_CHARS
        at_cap = deepcopy(minimal)
        at_cap["world_hints_for_background"] = [hint_at_cap]
        sbi_at = {
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": [hint_at_cap],
        }
        rep_pass = bsx.run_validation(
            world_brief=world, minimal_brief=at_cap, stage_b_input=sbi_at,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert rep_pass["passed"], (
            f"at-cap world hint must PASS: {rep_pass['failed_checks']!r}"
        )
        # over-cap world hint -> FAIL
        hint_over = "y" * (bsx.STAGE_B_WORLD_HINTS_ITEM_CHARS + 1)
        over = deepcopy(minimal)
        over["world_hints_for_background"] = [hint_over]
        sbi_over = {
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": [hint_over],
        }
        rep_fail = bsx.run_validation(
            world_brief=world, minimal_brief=over, stage_b_input=sbi_over,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert not rep_fail["passed"]
        assert any("cap_exceeded" in f and "world_hints_item" in f
                   for f in rep_fail["failed_checks"])
        # over-cap continuity_groups items count -> FAIL
        too_many_groups = deepcopy(minimal)
        too_many_groups["continuity_groups"] = [
            {"group_id": f"cg_{i}", "label": "x",
             "what_must_stay_consistent": [], "allowed_variations": [],
             "evidence_refs": [
                 {"source_ref": "planning_doc:test",
                  "quote": "coastal village at present day",
                  "confidence": "plausible"},
             ]}
            for i in range(bsx.STAGE_B_CONTINUITY_GROUPS_MAX + 1)
        ]
        rep3 = bsx.run_validation(
            world_brief=world, minimal_brief=too_many_groups,
            stage_b_input=sbi, bundle=bundle, spec=spec,
            world_schema=ws, minimal_schema=ms,
        )
        assert any("cap_exceeded" in f and "continuity_groups_items" in f
                   for f in rep3["failed_checks"])

    def test_checker_enforces_stage_b_input_contract(self):
        """6-D-21 (W0g BLOCKING): structural input contract only.
        - Stage B input top-level must contain exactly {world_brief_ref,
          world_hints_for_background} as world keys.
        - No `world_background_brief` embedded.
        - No `irrelevant_or_do_not_pass_down`.
        """
        spec, bundle = _minimal_synthetic_bundle()
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        world, minimal, _ = _basic_valid_briefs(spec, bundle)
        # leaked: full world dict embedded
        leaked = {
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": ["x"],
            "world_background_brief": {"era_and_time_period": "p"},
        }
        rep = bsx.run_validation(
            world_brief=world, minimal_brief=minimal, stage_b_input=leaked,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert any("stage_b_input_contract" in f
                   and "world_background_brief_embedded" in f
                   for f in rep["failed_checks"])
        # leaked: irrelevant list
        leaked2 = {
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": ["x"],
            "irrelevant_or_do_not_pass_down": ["plot lore"],
        }
        rep2 = bsx.run_validation(
            world_brief=world, minimal_brief=minimal, stage_b_input=leaked2,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert any("stage_b_input_contract" in f
                   and "irrelevant_list_in_stage_b_input" in f
                   for f in rep2["failed_checks"])
        # leaked: extra world_ key
        leaked3 = {
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": ["x"],
            "world_extra_key": "anything",
        }
        rep3 = bsx.run_validation(
            world_brief=world, minimal_brief=minimal, stage_b_input=leaked3,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert any("stage_b_input_contract" in f
                   and "world_keys_must_be_exactly" in f
                   for f in rep3["failed_checks"])


# ---------------------------------------------------------------------------
# 6-E. Prompt / CLI / production / output (2)
# ---------------------------------------------------------------------------
class TestPromptCliOutput:
    def test_prompts_separate_world_and_spatial_and_spatial_builder_excludes_full_stage_a(self):
        """6-E-22 (W0g BLOCKING 3): file presence + spatial prompt builder
        signature has no world_background_brief / irrelevant_or_do_not_pass_down
        params. spatial_user.txt body content NOT grep'd (W0g leakage avoidance).
        """
        # Function signature checks
        sb_sig = inspect.signature(bsx.build_prompt_spatial_user)
        param_names = set(sb_sig.parameters.keys())
        assert "world_background_brief" not in param_names, (
            f"spatial prompt builder must NOT accept world_background_brief; "
            f"params={param_names}"
        )
        assert "irrelevant_or_do_not_pass_down" not in param_names, (
            f"spatial prompt builder must NOT accept "
            f"irrelevant_or_do_not_pass_down; params={param_names}"
        )
        assert "world_brief_ref" in param_names
        assert "world_hints_for_background" in param_names

        # File presence — dry-run smoke
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        ph_pack = bsx.placeholder_evidence_pack(bundle)
        # All 4 downstream prompt builders must produce non-empty strings,
        # receiving only the filtered evidence_pack (W2a).
        psys = bsx.build_prompt_world_system(spec)
        puser = bsx.build_prompt_world_user(
            spec=spec, evidence_pack=ph_pack, schema=ws,
        )
        ssys = bsx.build_prompt_spatial_system(spec)
        suser = bsx.build_prompt_spatial_user(
            spec=spec, evidence_pack=ph_pack, schema=ms,
            world_brief_ref="world_background_brief",
            world_hints_for_background=["short hint"],
        )
        for body in (psys, puser, ssys, suser):
            assert isinstance(body, str) and body.strip()

    def test_cli_generate_default_off_and_no_network_no_db_write_and_outputs_18(
            self, monkeypatch, tmp_path):
        """6-E-23 combined CLI + production + output integrity (W2a updates
        the file count from 13 → 18).
        - --generate default off (LLM call 0)
        - --dry-run + --generate => generate False
        - backend/app diff empty
        - no DB write patterns
        - no network calls during dry-run (monkeypatched)
        - 18 output files emitted (added: evidence_filter schema,
          evidence_filter_system/user prompts, evidence_pack,
          filter_validation_report)
        """
        # CLI default
        args1 = bsx.parse_args([])
        assert args1.generate is False

        # --dry-run forces generate off (resolved by main)
        spec = bsx.build_sample_fixture_l05_spec()
        args2 = bsx.parse_args(["--dry-run", "--generate"])
        eff = bsx._resolve_effective_args(args2, spec)
        assert eff["generate"] is False

        # backend/app diff empty
        result = subprocess.run(
            ["git", "diff", "--stat", "HEAD", "--",
             "backend/app", "backend/alembic"],
            cwd=str(_REPO_ROOT), capture_output=True, text=True,
            check=False,
        )
        assert result.returncode == 0
        assert not result.stdout.strip(), (
            f"backend/app or backend/alembic has uncommitted diff: "
            f"{result.stdout!r}"
        )

        # No DB write patterns in script body
        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"script has forbidden DB-write pattern: {pat!r}"
            )

        # Monkeypatch network entry points before dry-run smoke
        called: list[str] = []
        def _trap(name):
            def _inner(*a, **kw):
                called.append(name)
                raise AssertionError(f"network call attempted: {name}")
            return _inner
        monkeypatch.setattr(_urlreq, "urlopen", _trap("urllib.urlopen"))
        monkeypatch.setattr(socket, "create_connection",
                            _trap("socket.create_connection"))

        # Dry-run via write_outputs (bypass DB by hand-building bundle)
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="planning", episode_text="episode",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        ph_pack = bsx.placeholder_evidence_pack(bundle)
        ph_world = {"placeholder": True, "stage": "world"}
        ph_min = {"placeholder": True, "stage": "spatial"}
        report = bsx.planned_checker_report()
        filter_rep = bsx.planned_evidence_filter_report()
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": False, "dry_run": True},
            outputs=[], model=spec.default_model,
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=report, run_meta=run_meta,
            include_diagnostic=False,
        )
        out_dir = tmp_path / "run"
        bsx.write_outputs(
            run_dir=out_dir, bundle=bundle,
            world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            world_brief=ph_world, minimal_brief=ph_min,
            validation_report=report, html=html, run_meta=run_meta,
        )
        assert not called, f"network call attempted: {called!r}"

        # 18 output files (W2a adds filter schema + 2 prompts + pack +
        # filter validation report on top of the original 13).
        expected_files = [
            "source_bundle.json", "source_bundle.md",
            "background_evidence_filter_schema.json",
            "world_brief_schema.json", "minimal_spatial_brief_schema.json",
            "prompt/evidence_filter_system.txt",
            "prompt/evidence_filter_user.txt",
            "prompt/world_system.txt", "prompt/world_user.txt",
            "prompt/spatial_system.txt", "prompt/spatial_user.txt",
            "background_evidence_pack.json",
            "world_brief.json", "minimal_spatial_brief.json",
            "filter_validation_report.json",
            "validation_report.json", "run_meta.json", "index.html",
        ]
        for rel in expected_files:
            assert (out_dir / rel).exists(), f"missing output file: {rel}"

        # HTML markers (W2-simple wording).
        html_body = (out_dir / "index.html").read_text(encoding="utf-8")
        for marker in ["backgroundcontinuitybrief", "evidence filter",
                        "raw sourcebundle"]:
            assert marker.lower() in html_body.lower(), (
                f"HTML missing marker: {marker!r}"
            )


# ---------------------------------------------------------------------------
# 6-F. No-truncate / no-auto-shorten guards (W1b — Codex 2026-05-24 사용자 추가 standing rule).
# cap 초과는 checker validation fail 로만 처리. 원문 보존. truncate / slice /
# drop / chunk / auto-shorten 0.
# ---------------------------------------------------------------------------
class TestNoTruncateGuards:
    def test_checker_preserves_over_cap_brief_content_verbatim(self):
        """6-F-24: over-cap brief 가 checker 통과 후 brief content 변경 0.
        checker 는 fail 만 기록, content 수정 X.
        """
        spec, bundle = _minimal_synthetic_bundle()
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        world, minimal, sbi = _basic_valid_briefs(spec, bundle)
        # over-cap world hint
        long_hint = "z" * (bsx.STAGE_B_WORLD_HINTS_ITEM_CHARS + 50)
        minimal["world_hints_for_background"] = [long_hint]
        sbi_over = {
            "world_brief_ref": "world_background_brief",
            "world_hints_for_background": [long_hint],
        }
        # Run checker
        bsx.run_validation(
            world_brief=world, minimal_brief=minimal, stage_b_input=sbi_over,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        # Brief content must remain verbatim — same length, same string.
        assert minimal["world_hints_for_background"] == [long_hint], (
            "checker auto-truncated the world hint — must preserve original"
        )
        assert len(minimal["world_hints_for_background"][0]) == len(long_hint)
        # Verify over-cap was indeed a fail (so we know we exercised the cap
        # path).
        rep = bsx.run_validation(
            world_brief=world, minimal_brief=minimal, stage_b_input=sbi_over,
            bundle=bundle, spec=spec, world_schema=ws, minimal_schema=ms,
        )
        assert any("cap_exceeded" in f for f in rep["failed_checks"])

    def test_source_bundle_preserves_long_input_text_verbatim(self):
        """6-F-25: synthetic 매우 긴 planning_doc text 가 source_bundle.text
        에 truncate 없이 그대로. truncate/slice/drop/chunk 0.
        """
        spec = bsx.build_sample_fixture_l05_spec()
        long_planning = "A" * 50_000
        long_episode = "B" * 30_000
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text=long_planning, episode_text=long_episode,
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        planning_src = next(
            s for s in bundle.sources if s.kind == "planning_doc"
        )
        episode_src = next(
            s for s in bundle.sources if s.kind == "episode_fulltext"
        )
        # NFC normalize is identity on pure-ASCII input, so length must match.
        assert planning_src.text == long_planning, (
            "build_source_bundle truncated planning_doc"
        )
        assert planning_src.char_count == len(long_planning)
        assert episode_src.text == long_episode
        assert episode_src.char_count == len(long_episode)

    def test_script_has_no_truncation_logic_for_brief_or_source(self):
        """6-F-26: script source AST 에 brief item field / source.text 를
        cap 적용 슬라이스로 자르는 패턴 0. 허용되는 slice:
          - HTML preview (src.text[:200], src.sha256[:16], prompt[:600])
          - dir-name format check (name[:8].isdigit())
          - id generation (uuid.uuid4().hex[:6])
          - schema body 의 substring/regex check 자체 0
        금지: brief field 또는 source.text 의 *원문 보존을 깨는* slice/
        truncate.
        """
        tree = _module_ast()
        # Walk every Subscript with a Slice — examine the value being sliced.
        forbidden_source_targets = (
            "world_hints_for_background", "place_identity",
            "continuity_groups", "essential_spatial_relations",
            "visual_anchors", "state_variations",
            "generation_notes", "world_brief_ref",
            "world_background_brief", "irrelevant_or_do_not_pass_down",
        )
        for sub in ast.walk(tree):
            if isinstance(sub, ast.Subscript) and isinstance(sub.slice, ast.Slice):
                value_src = ast.unparse(sub.value)
                # No slice may target a brief content field name.
                for target in forbidden_source_targets:
                    assert target not in value_src, (
                        f"forbidden truncation: {ast.unparse(sub)!r} "
                        f"slices brief field {target!r}"
                    )
        # Also assert no obvious truncate helpers imported / used.
        body = _module_source()
        for pat in ("textwrap.shorten", ".truncate(", "[:cap]", "[:MAX_",
                     "[:max_"):
            assert pat not in body, (
                f"forbidden truncation pattern in script: {pat!r}"
            )

    def test_source_loaders_do_not_strip_or_normalize_text(self):
        """6-F-27 (W1c BLOCKING): DB loaders + _make_source_item must NOT
        call .strip / .lstrip / .rstrip / unicodedata.normalize on source
        text. Only `.env` parser (`_load_backend_env`) is allowed to use
        .strip on its key/value lines (env config, not source content).
        """
        tree = _module_ast()
        loader_names = {
            "load_planning_doc", "load_episode_fulltext",
            "load_selected_shots_for_location", "load_entity_catalog",
            "load_location_catalog", "_make_source_item",
            "build_source_bundle",
        }
        forbidden_methods = {"strip", "lstrip", "rstrip"}
        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 in forbidden_methods):
                    pytest.fail(
                        f"source loader {node.name!r} calls forbidden "
                        f".{sub.func.attr}(): {ast.unparse(sub)}"
                    )
        # `unicodedata.normalize` 0 — entire script body (W1c verbatim).
        body = _module_source()
        assert "unicodedata.normalize" not in body, (
            "script must not call unicodedata.normalize on source text"
        )
        assert "import unicodedata" not in body, (
            "script must not import unicodedata (verbatim source preservation)"
        )

    def test_source_bundle_preserves_leading_trailing_whitespace_verbatim(self):
        """6-F-28 (W1c BLOCKING): build_source_bundle preserves leading /
        trailing whitespace, newlines, tabs in source text. char_count ==
        len(original_text).
        """
        spec = bsx.build_sample_fixture_l05_spec()
        weird_plan = "\n\n  \t leading and trailing whitespace planning   \n  "
        weird_ep = "  \t episode body with tabs and newlines\n  \n"
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text=weird_plan, episode_text=weird_ep,
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        planning_src = next(
            s for s in bundle.sources if s.kind == "planning_doc"
        )
        episode_src = next(
            s for s in bundle.sources if s.kind == "episode_fulltext"
        )
        assert planning_src.text == weird_plan, (
            f"planning text mutated: {planning_src.text!r} != {weird_plan!r}"
        )
        assert planning_src.char_count == len(weird_plan)
        assert episode_src.text == weird_ep
        assert episode_src.char_count == len(weird_ep)


# ---------------------------------------------------------------------------
# 6-G. BackgroundEvidencePack (Evidence Filter stage, W2a — Codex 2026-05-24)
#
# 79 sources of raw SourceBundle is read ONLY by the Evidence Filter; the
# downstream world / spatial prompt builders receive only the filtered
# BackgroundEvidencePack. This group enforces the structural contract.
# ---------------------------------------------------------------------------
def _evidence_pack_basic(bundle):
    """W2-simple BackgroundContinuityBrief — minimal valid brief that
    passes the deterministic checker (synthetic single-planning-doc
    bundle). Codex Q3: evidence_refs carry provenance ONLY (no
    confidence_band — that sits on the rule)."""
    er = {
        "source_ref": "planning_doc:test",
        "quote": "coastal village at present day",
    }
    return {
        "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
        "source_bundle_ref": bundle.bundle_id,
        "common_place_identity": {
            "summary":
                "A small place that recurs across scenes as a stable "
                "baseline.",
            "why_this_identity":
                "The source describes this location as the shared "
                "place anchor.",
            "confidence_band": "plausible",
            "evidence_refs": [er],
        },
        "must_stay_consistent": [{
            "rule_id": "scale_baseline",
            "statement": "Keep the baseline scale and material feel.",
            "why_consistent": "Source anchors the place description.",
            "confidence_band": "plausible",
            "evidence_refs": [er],
        }],
        "must_not_contradict": [{
            "rule_id": "no_modern_luxury",
            "do_not_introduce": "luxury or futuristic surfaces.",
            "why": "source describes a present-day setting.",
            "confidence_band": "plausible",
            "evidence_refs": [er],
        }],
        "allowed_creative_freedom": [{
            "rule_id": "minor_props",
            "free_to_choose": "wall props and small decor.",
            "guidance": "must remain consistent with baseline scale.",
            "basis": "art_direction_choice",
            "confidence_band": "plausible",
            "evidence_refs": [],
        }],
        "state_change_rules": [{
            "rule_id": "lighting_shift",
            "change_kind": "lighting variation",
            "when_applies": "different times of day.",
            "what_changes": ["overall warmth and shadow length"],
            "what_stays": ["spatial layout"],
            "confidence_band": "plausible",
            "evidence_refs": [er],
        }],
        # No shot_conflict_checks in this minimal fixture — the
        # synthetic bundle has no selected_shot source. Tests that need
        # to exercise the per-shot grounding contract build a
        # bundle-with-shot explicitly.
        "shot_conflict_checks": [],
        "unknowns_left_to_art_direction": [{
            "area": "specific furniture",
            "note": "not specified by source.",
        }],
        "coverage_notes": [{
            "note": "no info on weather",
            "consequence_for_background": "keep weather loose",
            "confidence_band": "weak",
        }],
        "rejected_or_irrelevant_summary": [{
            "short_note": "character drama details",
            "why_not_pass_down": "not visual background.",
        }],
    }


def _continuity_brief_with_one_bad_quote(bundle):
    """ContinuityBrief whose only failure is one evidence_ref quote not
    present in the source (correction-trigger fixture)."""
    er_good = {
        "source_ref": "planning_doc:test",
        "quote": "coastal village at present day",
    }
    er_bad = {
        "source_ref": "planning_doc:test",
        "quote": "this paraphrase is not in the source text",
    }
    pack = _evidence_pack_basic(bundle)
    # Inject the bad evidence into one rule.
    pack["must_stay_consistent"][0]["evidence_refs"] = [er_good, er_bad]
    return pack


def _continuity_brief_with_enum_only_failure(bundle):
    """ContinuityBrief whose only failure is a rule-level
    confidence_band outside the enum — correction must NOT trigger."""
    pack = _evidence_pack_basic(bundle)
    pack["must_stay_consistent"][0]["confidence_band"] = "high"
    return pack


class TestEvidenceFilterSchemaShape:
    """6-G (W2-simple refresh): BackgroundContinuityBrief schema +
    common_place_identity required + caps + legacy-keys forbidden."""

    def test_brief_schema_has_required_top_level_keys(self):
        _, bundle = _minimal_synthetic_bundle()
        s = bsx.build_background_evidence_pack_schema(bundle)
        for k in [
            "schema_version", "source_bundle_ref",
            "common_place_identity",
            "must_stay_consistent", "must_not_contradict",
            "allowed_creative_freedom", "state_change_rules",
            "shot_conflict_checks",
            "unknowns_left_to_art_direction",
            "coverage_notes", "rejected_or_irrelevant_summary",
        ]:
            assert k in s.get("required", []), k

    def test_brief_common_place_identity_required_fields(self):
        _, bundle = _minimal_synthetic_bundle()
        s = bsx.build_background_evidence_pack_schema(bundle)
        cpi = s["common_place_identity"]
        for k in ("summary", "why_this_identity",
                   "confidence_band", "evidence_refs"):
            assert k in cpi["required"], k
        assert cpi["properties"]["evidence_refs"]["minItems"] == 1

    def test_brief_allowed_creative_freedom_basis_enum_and_optional_evidence(self):
        _, bundle = _minimal_synthetic_bundle()
        s = bsx.build_background_evidence_pack_schema(bundle)
        item = s["allowed_creative_freedom"]["items"]
        # `basis` is required (enum); evidence_refs are NOT in
        # required (optional).
        assert "basis" in item["required"]
        assert "evidence_refs" not in item["required"]
        assert set(s["allowed_creative_freedom_basis_enum"]) == {
            "not_specified", "weakly_constrained",
            "art_direction_choice",
        }

    def test_brief_rule_sections_carry_rule_level_confidence_band(self):
        _, bundle = _minimal_synthetic_bundle()
        s = bsx.build_background_evidence_pack_schema(bundle)
        for section in (
            "must_stay_consistent", "must_not_contradict",
            "allowed_creative_freedom", "state_change_rules",
            "shot_conflict_checks",
        ):
            item = s[section]["items"]
            assert "confidence_band" in item["required"], section
            assert item["properties"]["confidence_band"]["enum"] == [
                "trusted", "plausible", "weak", "unknown",
            ]

    def test_brief_evidence_ref_shape_has_no_confidence_band(self):
        """Codex Q3: evidence_ref carries provenance only (source_ref +
        quote + optional positions). confidence sits on the rule."""
        er = bsx._brief_evidence_ref_schema()
        assert set(er["required"]) == {"source_ref", "quote"}
        assert "confidence_band" not in er["properties"]

    def test_brief_schema_forbids_legacy_and_inventory_and_topology_keys(self):
        _, bundle = _minimal_synthetic_bundle()
        s = bsx.build_background_evidence_pack_schema(bundle)
        forbidden = s.get("forbidden_top_level_or_item_fields", [])
        for k in ("count", "door_count", "window_count",
                  "furniture_count", "object_count", "inventory"):
            assert k in forbidden, k
        for k in bsx.TOPOLOGY_FIELDS_FORBIDDEN:
            assert k in forbidden, k
        for k in ("use_for", "pass_down_hint", "why_background_relevant",
                  "role", "applies_to", "background_fact", "why_keep",
                  "evidence_items"):
            assert k in forbidden, (
                f"legacy key {k!r} must be forbidden in W2-simple briefs"
            )

    def test_brief_schema_caps_present(self):
        _, bundle = _minimal_synthetic_bundle()
        s = bsx.build_background_evidence_pack_schema(bundle)
        caps = s["caps"]
        assert caps["list_max_items"] == 8
        assert caps["shot_conflict_checks_max_items"] == 24
        assert caps["statement_max_chars"] == 200
        assert caps["why_max_chars"] == 180
        assert caps["identity_summary_max_chars"] == 240
        assert caps["coverage_notes_max_items"] == 8
        assert caps["rejected_or_irrelevant_summary_max_items"] == 8


class TestEvidenceFilterChecker:
    """6-G (W2-simple refresh): checker — provenance + shape + cap +
    rule-level confidence + verbatim preservation."""

    def test_checker_validates_basic_pack_passes(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        # The basic pack uses shot_ref="shot:placeholder" which is not a
        # real selected_shot in the synthetic bundle; remove the
        # shot_conflict_checks section to keep this baseline test
        # focused on schema/provenance rather than shot equality.
        pack["shot_conflict_checks"] = []
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert rep["passed"], rep["failed_checks"]

    def test_checker_quote_not_in_source_fails(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        pack["must_stay_consistent"][0]["evidence_refs"][0]["quote"] = (
            "this quote is absent"
        )
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("quote_not_in_source" in f
                   for f in rep["failed_checks"])

    def test_checker_source_ref_resolvable(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        pack["must_stay_consistent"][0]["evidence_refs"][0]["source_ref"] = (
            "unknown:xyz"
        )
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("source_ref_unresolved" in f
                   for f in rep["failed_checks"])

    def test_checker_rejects_evidence_ref_with_confidence_band(self):
        """Codex Q3: evidence_ref carrying confidence_band is rejected
        — confidence sits on the rule."""
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        pack["must_stay_consistent"][0]["evidence_refs"][0][
            "confidence_band"] = "trusted"
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any(
            "evidence_ref_has_forbidden_confidence_band" in f
            for f in rep["failed_checks"]
        )

    def test_checker_rejects_rule_confidence_band_outside_enum(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        pack["must_stay_consistent"][0]["confidence_band"] = "high"
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("rule_confidence_invalid" in f
                   for f in rep["failed_checks"])

    def test_checker_rejects_allowed_creative_freedom_basis_outside_enum(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        pack["allowed_creative_freedom"][0]["basis"] = "freeform_label"
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any(
            "allowed_creative_freedom_basis_invalid" in f
            for f in rep["failed_checks"]
        )

    def test_checker_rule_id_must_be_slug(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        pack["must_stay_consistent"][0]["rule_id"] = "bad id with spaces!"
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("rule_id_not_slug" in f
                   for f in rep["failed_checks"])

    def test_checker_enforces_caps_and_preserves_over_cap_verbatim(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        long_statement = "z" * (bsx.CONTINUITY_BRIEF_STATEMENT_CHARS + 20)
        pack["must_stay_consistent"][0]["statement"] = long_statement
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("cap_exceeded" in f and "statement" in f
                   for f in rep["failed_checks"])
        # Verbatim preservation — checker must not mutate the brief.
        assert pack["must_stay_consistent"][0]["statement"] == long_statement

    def test_checker_rejects_inventory_or_count_keys(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        bad = deepcopy(pack)
        bad["must_stay_consistent"][0]["door_count"] = 2
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=bad, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("forbidden_key" in f and "door_count" in f
                   for f in rep["failed_checks"])

    def test_checker_over_cap_list_items_fails(self):
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        rule = pack["must_stay_consistent"][0]
        pack["must_stay_consistent"] = [
            {**rule, "rule_id": f"r_{i}"}
            for i in range(bsx.CONTINUITY_BRIEF_LIST_MAX + 1)
        ]
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("cap_exceeded" in f
                   and "must_stay_consistent_items" in f
                   for f in rep["failed_checks"])

    def test_checker_rejects_shot_ref_not_in_selected_shots(self):
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="coastal village at present day",
            episode_text="ep body",
            selected_shots=[{
                "still_id": "abc-123",
                "scene_index": 1, "shot_index": 1,
                "shot_description": "x", "scene_summary": "y",
                "visible_entities_json": "[]",
                "loc_short_ids_visible": [],
                "is_selected": True, "status": None,
            }],
            entity_catalog=[], location_catalog=[],
        )
        pack = _evidence_pack_basic(bundle)
        # Rewrite evidence refs to use the real planning_doc source_ref.
        pl_ref = next(
            s.source_ref for s in bundle.sources if s.kind == "planning_doc"
        )
        for section in ("must_stay_consistent", "must_not_contradict",
                        "allowed_creative_freedom",
                        "state_change_rules",
                        "shot_conflict_checks"):
            for it in pack[section]:
                if isinstance(it, dict):
                    for er in (it.get("evidence_refs") or []):
                        er["source_ref"] = pl_ref
        cpi = pack["common_place_identity"]
        for er in cpi["evidence_refs"]:
            er["source_ref"] = pl_ref
        pack["source_bundle_ref"] = bundle.bundle_id
        # Inject one shot_conflict_check with a bogus shot_ref to
        # trigger the selected_shots exact-equality fail.
        pack["shot_conflict_checks"] = [{
            "shot_ref": "shot:nonexistent",
            "must_support": ["x"],
            "avoid_contradiction": ["y"],
            "confidence_band": "plausible",
            "evidence_refs": [{
                "source_ref": pl_ref,
                "quote": "coastal village at present day",
            }],
        }]
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep["passed"]
        assert any("shot_ref_not_in_selected_shots" in f
                   for f in rep["failed_checks"])


class TestEvidenceFilterPromptContract:
    """6-G-39..42: prompt builder structural contract.

    - The evidence_filter prompt builder receives the SourceBundle in full
      (this is the ONE function that consumes raw sources).
    - The world / spatial prompt builders MUST NOT receive a SourceBundle
      or any raw `source.text`; they only see the filtered EvidencePack.
    """

    def test_evidence_filter_prompt_builder_signature_accepts_bundle(self):
        sig = inspect.signature(bsx.build_prompt_evidence_filter_user)
        params = set(sig.parameters.keys())
        assert "bundle" in params, (
            f"evidence filter user prompt builder must accept bundle: "
            f"{params}"
        )
        assert "schema" in params

    def test_evidence_filter_system_and_user_builders_present(self):
        # Both functions exist and return non-empty strings.
        spec = bsx.build_sample_fixture_l05_spec()
        _, bundle = _minimal_synthetic_bundle()
        sschema = bsx.build_background_evidence_pack_schema(bundle)
        sys_txt = bsx.build_prompt_evidence_filter_system(spec)
        usr_txt = bsx.build_prompt_evidence_filter_user(
            spec=spec, bundle=bundle, schema=sschema,
        )
        assert isinstance(sys_txt, str) and sys_txt.strip()
        assert isinstance(usr_txt, str) and usr_txt.strip()

    def test_world_prompt_builder_does_not_accept_bundle_or_raw_sources(self):
        """world prompt builder takes the EvidencePack only — no bundle,
        no `selected_shots`, no raw `source.text`."""
        sig = inspect.signature(bsx.build_prompt_world_user)
        params = set(sig.parameters.keys())
        assert "bundle" not in params, (
            f"world prompt builder must NOT accept bundle anymore; "
            f"params={params}"
        )
        for forbidden in ("source_bundle", "selected_shots",
                          "planning_text", "episode_text"):
            assert forbidden not in params, (
                f"world prompt builder must NOT accept raw source param "
                f"{forbidden!r}; params={params}"
            )
        assert "evidence_pack" in params, (
            f"world prompt builder must accept evidence_pack; params={params}"
        )

    def test_spatial_prompt_builder_does_not_accept_bundle_or_raw_sources(self):
        sig = inspect.signature(bsx.build_prompt_spatial_user)
        params = set(sig.parameters.keys())
        assert "bundle" not in params, (
            f"spatial prompt builder must NOT accept bundle anymore; "
            f"params={params}"
        )
        for forbidden in ("source_bundle", "selected_shots",
                          "planning_text", "episode_text",
                          "world_background_brief",
                          "irrelevant_or_do_not_pass_down"):
            assert forbidden not in params, (
                f"spatial prompt builder must NOT accept raw/leak param "
                f"{forbidden!r}; params={params}"
            )
        assert "evidence_pack" in params
        assert "world_brief_ref" in params
        assert "world_hints_for_background" in params


class TestEvidenceFilterRunIntegration:
    """6-G-43..47: dry-run placeholder, --generate failed status, output
    files, HTML explains filter, run_meta stage list."""

    def test_dry_run_emits_placeholder_evidence_pack_and_filter_plan(
            self, tmp_path):
        """Dry-run wires the placeholder pack into the output dir; the
        validation report is `planned only` (executed=False) and contains
        Evidence Filter planned checks."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="planning", episode_text="episode",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        ph_pack = bsx.placeholder_evidence_pack(bundle)
        ph_world = {"placeholder": True, "stage": "world"}
        ph_min = {"placeholder": True, "stage": "spatial"}
        rep = bsx.planned_checker_report()
        filter_rep = bsx.planned_evidence_filter_report()
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": False, "dry_run": True},
            outputs=[], model=spec.default_model,
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=rep, run_meta=run_meta,
            include_diagnostic=False,
        )
        out_dir = tmp_path / "run"
        bsx.write_outputs(
            run_dir=out_dir, bundle=bundle,
            world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            world_brief=ph_world, minimal_brief=ph_min,
            validation_report=rep, html=html, run_meta=run_meta,
        )
        # Placeholder pack written
        pack_file = out_dir / "background_evidence_pack.json"
        assert pack_file.exists()
        loaded = json.loads(pack_file.read_text(encoding="utf-8"))
        assert loaded.get("placeholder") is True
        assert loaded.get("source_bundle_ref") == bundle.bundle_id
        # Filter schema written
        assert (out_dir / "background_evidence_filter_schema.json").exists()
        # Filter prompt pair written
        assert (out_dir / "prompt" / "evidence_filter_system.txt").exists()
        assert (out_dir / "prompt" / "evidence_filter_user.txt").exists()
        # Filter validation report written and is planned (executed=False)
        fv = json.loads(
            (out_dir / "filter_validation_report.json").read_text(
                encoding="utf-8"))
        assert fv.get("executed") is False
        assert "planned_checks" in fv

    def test_evidence_filter_run_generate_failed_status_when_api_unavailable(
            self, monkeypatch):
        """When --generate is on but the model client cannot be reached
        (missing key or network failure), run_evidence_filter_generate
        returns a clear failed status with retry budget recorded.
        Never raises; never infinite-loops."""
        _, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)

        def _always_fail(*a, **kw):
            raise RuntimeError("api unreachable in test")
        result = bsx.run_evidence_filter_generate(
            bundle=bundle, schema=schema,
            model="gemini-3.5-flash",
            llm_call=_always_fail,
            max_retries=1,
        )
        assert result["status"] == "failed"
        assert result["attempts"] == 2  # initial + 1 retry
        assert "api unreachable" in result.get("error", "")
        # No partial pack written; only a recorded reason.
        assert result.get("evidence_pack") is None

    def test_run_meta_lists_background_continuity_brief_stage_first(self):
        """W2-simple-b: the first stage is the BackgroundContinuityBrief
        (world_brief / minimal_spatial_brief are legacy placeholders)."""
        spec = bsx.build_sample_fixture_l05_spec()
        rm = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": False, "dry_run": True},
            outputs=[], model=spec.default_model,
            char_budget_estimate=0, bundle_sha256="x" * 64,
        )
        stages = rm.get("stages", [])
        assert stages[0] == "background_continuity_brief"
        # Legacy placeholders may still appear but must be marked.
        assert any(s.startswith("_legacy_") for s in stages[1:])

    def test_html_explains_evidence_filter_role(self, tmp_path):
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        ph_pack = bsx.placeholder_evidence_pack(bundle)
        rep = bsx.planned_checker_report()
        filter_rep = bsx.planned_evidence_filter_report()
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": False, "dry_run": True},
            outputs=[], model=spec.default_model,
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=rep, run_meta=run_meta,
            include_diagnostic=False,
        )
        low = html.lower()
        # Required explanatory phrase (exact wording in spec, in HTML body).
        assert ("full sourcebundle was read only by the evidence filter"
                in low), html[:2000]
        assert "downstream background briefs use only filtered evidence" in low

    def test_outputs_now_include_filter_artifacts_18_total(
            self, tmp_path, monkeypatch):
        """Combined output integrity: 18 output files when the pack is
        valid (13 original + evidence_filter_schema + 2 filter prompts +
        evidence_pack + filter_validation_report). On `validation_failed`
        an additional 19th file `background_evidence_pack_quarantined.json`
        is written (covered by `TestGenerateFailClosed`). HTML explains
        the Evidence Filter role."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="planning", episode_text="episode",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        ph_pack = bsx.placeholder_evidence_pack(bundle)
        ph_world = {"placeholder": True, "stage": "world"}
        ph_min = {"placeholder": True, "stage": "spatial"}
        report = bsx.planned_checker_report()
        filter_rep = bsx.planned_evidence_filter_report()
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": False, "dry_run": True},
            outputs=[], model=spec.default_model,
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=report, run_meta=run_meta,
            include_diagnostic=False,
        )
        out_dir = tmp_path / "run"
        bsx.write_outputs(
            run_dir=out_dir, bundle=bundle,
            world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            world_brief=ph_world, minimal_brief=ph_min,
            validation_report=report, html=html, run_meta=run_meta,
        )
        expected_files = [
            "source_bundle.json", "source_bundle.md",
            "background_evidence_filter_schema.json",
            "world_brief_schema.json", "minimal_spatial_brief_schema.json",
            "prompt/evidence_filter_system.txt",
            "prompt/evidence_filter_user.txt",
            "prompt/world_system.txt", "prompt/world_user.txt",
            "prompt/spatial_system.txt", "prompt/spatial_user.txt",
            "background_evidence_pack.json",
            "world_brief.json", "minimal_spatial_brief.json",
            "filter_validation_report.json",
            "validation_report.json", "run_meta.json", "index.html",
        ]
        # 18 distinct outputs above (quarantine file is only written on
        # `validation_failed`, see TestGenerateFailClosed).
        for rel in expected_files:
            assert (out_dir / rel).exists(), f"missing: {rel}"


class TestEvidenceFilterStaticGuards:
    """6-G-48..49: static guards for the new code — Evidence Filter must
    not reintroduce lexicon matchers, classifier helpers, or any content
    scan; new checker function passes the existing 6-A AST audits."""

    def test_evidence_filter_checker_function_has_no_brief_content_scan(self):
        """Re-applies the 6-A "no content scan" guard to the Evidence
        Filter checker. Note: provenance via `source_text.find(quote)` is
        explicitly ALLOWED — that is exact-quote containment, not a
        content scan. What is forbidden is iterating over a deny list or
        calling `.contains` / `.lower()` / `.find` against pack content
        fields like why_background_relevant or pass_down_hint.
        """
        fn = _function_def_by_name("run_evidence_filter_validation")
        assert fn is not None, (
            "run_evidence_filter_validation function not found"
        )
        for sub in ast.walk(fn):
            if isinstance(sub, ast.For):
                if (isinstance(sub.iter, ast.Name)
                        and sub.iter.id.lower() in {
                            "deny_list", "denylist", "forbidden_words",
                            "blacklist", "bad_words",
                        }):
                    pytest.fail(
                        f"evidence filter checker iterates over deny-list "
                        f"name: {sub.iter.id}"
                    )
            if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute):
                if sub.func.attr in {"contains"}:
                    parent_src = ast.unparse(sub.func.value)
                    suspicious_targets = (
                        "why_background_relevant", "pass_down_hint",
                        "short_note", "consequence_for_background",
                    )
                    if any(t in parent_src for t in suspicious_targets):
                        pytest.fail(
                            f"evidence filter checker scans pack content via "
                            f".contains: {ast.unparse(sub)}"
                        )

    def test_evidence_filter_module_has_no_lexicon_constants(self):
        """The new evidence-filter code path must not add lexicon /
        classifier constants. Repeats the 6-A guard against any newly
        introduced module-level names."""
        forbidden_substrings = (
            "USE_FOR_LEXICON", "PASS_DOWN_TERMS",
            "EVIDENCE_TERM_LIST", "FILTER_LEXICON",
        )
        tree = _module_ast()
        for node in tree.body:
            if isinstance(node, ast.Assign):
                for tgt in node.targets:
                    if isinstance(tgt, ast.Name):
                        for ban in forbidden_substrings:
                            assert ban not in tgt.id, (
                                f"forbidden lexicon constant: {tgt.id}"
                            )


# ---------------------------------------------------------------------------
# 6-H. W2a-b BLOCKING / IMPORTANT patches (Codex review 2026-05-24).
#
# - BLOCKING 1: generate path's "succeeded API call" must NOT promote an
#   invalid EvidencePack into the downstream world/spatial prompt builders.
#   route_evidence_filter_result() decides downstream pack vs quarantine.
# - BLOCKING 2: evidence_filter prompts must spell out the confidence_band
#   enum (trusted/plausible/weak/unknown) and explicitly ban high/medium/low.
# - IMPORTANT 1: downstream world/spatial *system* prompts must say
#   "BackgroundEvidencePack", not "SourceBundle".
# ---------------------------------------------------------------------------
def _invalid_pack_with_high_confidence(bundle):
    """W2-simple BackgroundContinuityBrief whose only failure is a
    confidence_band outside the enum (`high`). Used to verify
    correction-skip behavior in fail-closed routing."""
    pack = _evidence_pack_basic(bundle)
    pack["common_place_identity"]["confidence_band"] = "high"  # invalid
    return pack


class TestGenerateFailClosed:
    """W2a-b BLOCKING 1: succeeded LLM call + failed validation must NOT
    silently promote the invalid pack to downstream prompts."""

    def test_route_succeeded_with_valid_validation_promotes_pack(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        valid_pack = _evidence_pack_basic(bundle)
        filter_result = {
            "status": "succeeded",
            "attempts": 1,
            "evidence_pack": valid_pack,
            "model": "gemini-3.5-flash",
            "error": "",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
        )
        assert routed["run_status"] == "succeeded"
        assert routed["exit_code"] == 0
        assert routed["downstream_pack"] is valid_pack
        assert routed["quarantined_pack"] is None
        assert routed["evidence_filter_report"]["passed"] is True

    def test_route_succeeded_with_failed_validation_quarantines_pack(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        invalid = _invalid_pack_with_high_confidence(bundle)
        filter_result = {
            "status": "succeeded",
            "attempts": 1,
            "evidence_pack": invalid,
            "model": "gemini-3.5-flash",
            "error": "",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
        )
        assert routed["run_status"] == "validation_failed"
        assert routed["exit_code"] != 0
        # Quarantine the invalid pack, hand a placeholder to downstream.
        assert routed["quarantined_pack"] is invalid
        assert routed["downstream_pack"].get("placeholder") is True
        assert routed["downstream_pack"]["source_bundle_ref"] == bundle.bundle_id
        report = routed["evidence_filter_report"]
        assert report["passed"] is False
        assert report["generate_status"] == "succeeded"
        # _invalid_pack_with_high_confidence puts the bad enum at the
        # common_place_identity level (W2-simple equivalent).
        assert any(
            "common_place_identity_confidence_invalid" in f
            for f in report["failed_checks"]
        )

    def test_route_generate_failed_uses_placeholder_and_nonzero_exit(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        filter_result = {
            "status": "failed",
            "attempts": 2,
            "evidence_pack": None,
            "model": "gemini-3.5-flash",
            "error": "api unreachable",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
        )
        assert routed["run_status"] == "generate_failed"
        assert routed["exit_code"] != 0
        assert routed["downstream_pack"].get("placeholder") is True
        assert routed["quarantined_pack"] is None

    def test_main_returns_nonzero_when_filter_validation_fails(
            self, monkeypatch, tmp_path):
        """End-to-end of the generate path with a stub LLM: invalid pack
        leads to non-zero exit code AND the quarantine artifact is
        written instead of polluting background_evidence_pack.json with
        invalid content."""
        spec = bsx.build_sample_fixture_l05_spec()

        # Stub the DB loaders so we never touch PostgreSQL.
        class _FakeSession:
            def __enter__(self):
                return self
            def __exit__(self, *a):
                return False
        def _fake_session_local():
            return _FakeSession()
        monkeypatch.setattr(
            "experiment_background_semantic_extractor."
            "_load_backend_env", lambda: None,
        )

        # Patch SessionLocal indirection via app.core.database import
        import app.core.database as _db
        monkeypatch.setattr(_db, "SessionLocal", _fake_session_local)

        # Patch loaders to return synthetic data.
        monkeypatch.setattr(bsx, "load_planning_doc",
                            lambda s, pid: ("planning", {"id": pid}))
        monkeypatch.setattr(bsx, "load_episode_fulltext",
                            lambda s, eid: ("episode", {"id": eid}))
        monkeypatch.setattr(
            bsx, "load_selected_shots_for_location",
            lambda s, pid, eid, lid: [],
        )
        monkeypatch.setattr(bsx, "load_entity_catalog",
                            lambda s, pid: [])
        monkeypatch.setattr(bsx, "load_location_catalog",
                            lambda s, pid: [])

        # Stub the LLM call: always return an invalid brief
        # (common_place_identity.confidence_band = "high").
        def _stub_llm(*, bundle, schema, model):
            er = {
                "source_ref": "planning_doc:" + spec.project_id,
                "quote": "planning",
            }
            return {
                "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
                "source_bundle_ref": bundle.bundle_id,
                "common_place_identity": {
                    "summary": "baseline",
                    "why_this_identity": "baseline",
                    "confidence_band": "high",  # invalid enum
                    "evidence_refs": [er],
                },
                "must_stay_consistent": [],
                "must_not_contradict": [],
                "allowed_creative_freedom": [],
                "state_change_rules": [],
                "shot_conflict_checks": [],
                "unknowns_left_to_art_direction": [],
                "coverage_notes": [],
                "rejected_or_irrelevant_summary": [],
            }
        monkeypatch.setattr(
            bsx, "_default_evidence_filter_llm_call", _stub_llm,
        )

        out_root = tmp_path / "runs"
        rc = bsx.main([
            "--generate", "--model", "gemini-3.5-flash",
            "--no-serve", "--output-root", str(out_root),
        ])
        # Non-zero exit because validation failed.
        assert rc != 0, "main must return non-zero on validation_failed"
        # Pick the only run dir created.
        run_dirs = list(out_root.iterdir())
        assert len(run_dirs) == 1
        run_dir = run_dirs[0]
        # background_evidence_pack.json now holds the *placeholder*, not
        # the invalid pack.
        pack = json.loads(
            (run_dir / "background_evidence_pack.json").read_text(
                encoding="utf-8"),
        )
        assert pack.get("placeholder") is True
        # The invalid pack is quarantined alongside the run.
        quarantine = run_dir / "background_evidence_pack_quarantined.json"
        assert quarantine.exists()
        q = json.loads(quarantine.read_text(encoding="utf-8"))
        assert q["common_place_identity"]["confidence_band"] == "high"
        # filter_validation_report records the validation_failed status.
        fv = json.loads(
            (run_dir / "filter_validation_report.json").read_text(
                encoding="utf-8"))
        assert fv["passed"] is False
        assert fv.get("run_status") == "validation_failed"
        assert fv.get("generate_status") == "succeeded"


class TestEvidenceFilterPromptEnumDiscipline:
    """W2a-b BLOCKING 2: prompts must spell out the confidence_band enum
    and explicitly ban high/medium/low."""

    def test_evidence_filter_system_prompt_enumerates_confidence_band(self):
        spec = bsx.build_sample_fixture_l05_spec()
        text = bsx.build_prompt_evidence_filter_system(spec)
        low = text.lower()
        for token in ("trusted", "plausible", "weak", "unknown"):
            assert token in low, (
                f"evidence_filter system prompt must mention "
                f"confidence enum token {token!r}"
            )
        for forbidden in ("high", "medium", "low"):
            # Either the prompt explicitly forbids the word, or it does
            # not mention it as a confidence label at all. Easiest signal:
            # the prompt mentions the forbidden value explicitly with a
            # negation context. Accept either pattern.
            assert forbidden in low, (
                f"evidence_filter system prompt should explicitly "
                f"mention {forbidden!r} (so the LLM is told NOT to use "
                f"it)"
            )
        assert "do not use" in low or "must not use" in low or "must be exactly one of" in low, (
            "system prompt should explicitly tell the LLM to avoid "
            "high/medium/low or enforce the enum exactly"
        )

    def test_evidence_filter_user_prompt_lists_confidence_enum(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        text = bsx.build_prompt_evidence_filter_user(
            spec=spec, bundle=bundle, schema=schema,
        )
        low = text.lower()
        for token in ("trusted", "plausible", "weak", "unknown"):
            assert token in low, (
                f"evidence_filter user prompt must enumerate "
                f"confidence_band value {token!r}"
            )


class TestDownstreamSystemPromptUsesPackNotBundle:
    """W2a-b IMPORTANT 1: world/spatial system prompts must speak about
    the filtered BackgroundEvidencePack, not the raw SourceBundle."""

    def test_world_system_prompt_says_pack_not_sourcebundle(self):
        spec = bsx.build_sample_fixture_l05_spec()
        text = bsx.build_prompt_world_system(spec)
        low = text.lower()
        assert "backgroundevidencepack" in low or "evidence pack" in low or "evidence-pack" in low, (
            "world system prompt must speak about the filtered "
            "BackgroundEvidencePack"
        )
        assert "read the provided sourcebundle" not in low, (
            "world system prompt must not instruct the LLM to read the "
            "raw SourceBundle (W2a contract)"
        )

    def test_spatial_system_prompt_says_pack_not_sourcebundle(self):
        spec = bsx.build_sample_fixture_l05_spec()
        text = bsx.build_prompt_spatial_system(spec)
        low = text.lower()
        assert "backgroundevidencepack" in low or "evidence pack" in low or "evidence-pack" in low, (
            "spatial system prompt must speak about the filtered "
            "BackgroundEvidencePack"
        )
        assert "read the provided sourcebundle" not in low, (
            "spatial system prompt must not instruct the LLM to read the "
            "raw SourceBundle (W2a contract)"
        )


# ---------------------------------------------------------------------------
# 6-I. W2a-c run_meta + HTML status honesty (Codex W2a-b review 2026-05-24).
#
# BLOCKING: filter_validation_report says validation_failed, but run_meta /
# HTML still look like a clean dry-run. Top-level metadata must surface
# the actual run status, exit code, and quarantine artifact so neither
# the user nor downstream code mistakes a fail-closed run for a success.
# IMPORTANT: PLAN_VERSION still says bsx_w1 even though W2a-b is a new
# contract (Evidence Filter gateway + fail-closed routing).
# ---------------------------------------------------------------------------
class TestRunMetaAndHtmlStatusHonesty:

    def test_build_run_meta_carries_run_status_and_exit_code(self):
        rm = bsx.build_run_meta(
            run_id="rt", fixture_id="fx",
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=10,
            bundle_sha256="x" * 64,
            run_status="validation_failed",
            exit_code=1,
            filter_validation_passed=False,
            generate_status="succeeded",
            filter_failed_checks=[
                "evidence_filter:quote_not_in_source:7",
            ],
            quarantine_output="background_evidence_pack_quarantined.json",
        )
        assert rm["run_status"] == "validation_failed"
        assert rm["exit_code"] == 1
        assert rm["filter_validation_passed"] is False
        assert rm["generate_status"] == "succeeded"
        assert rm["filter_failed_checks"] == [
            "evidence_filter:quote_not_in_source:7",
        ]
        assert rm["quarantine_output"] == "background_evidence_pack_quarantined.json"

    def test_build_run_meta_defaults_to_dry_run_when_status_omitted(self):
        rm = bsx.build_run_meta(
            run_id="rt", fixture_id="fx",
            args_dict={"generate": False}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=10,
            bundle_sha256="x" * 64,
        )
        assert rm["run_status"] == "dry_run"
        assert rm["exit_code"] == 0
        # dry-run does not run the checker, so passed must be None
        # (not False — False would mean executed-and-failed).
        assert rm["filter_validation_passed"] is None
        assert rm["quarantine_output"] is None

    def test_run_meta_lists_quarantine_when_validation_failed(
            self, tmp_path):
        """End-to-end via write_outputs: when quarantined_pack is provided,
        the file is written and run_meta.outputs lists it."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        invalid_pack = _invalid_pack_with_high_confidence(bundle)
        filter_rep = bsx.run_evidence_filter_validation(
            evidence_pack=invalid_pack, bundle=bundle, schema=fs,
        )
        filter_rep["run_status"] = "validation_failed"
        filter_rep["generate_status"] = "succeeded"
        placeholder_pack = {
            "placeholder": True, "stage": "evidence_filter",
            "source_bundle_ref": bundle.bundle_id,
            "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
            "reason": "validation failed; quarantined",
        }
        report = bsx.planned_checker_report()
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": True},
            outputs=["background_evidence_pack_quarantined.json"],
            model="gemini-3.5-flash",
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
            run_status="validation_failed",
            exit_code=1,
            filter_validation_passed=False,
            generate_status="succeeded",
            filter_failed_checks=filter_rep.get("failed_checks") or [],
            quarantine_output="background_evidence_pack_quarantined.json",
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=placeholder_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=report, run_meta=run_meta,
            include_diagnostic=False,
        )
        out_dir = tmp_path / "run"
        bsx.write_outputs(
            run_dir=out_dir, bundle=bundle,
            world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=placeholder_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            world_brief={}, minimal_brief={},
            validation_report=report, html=html, run_meta=run_meta,
            quarantined_pack=invalid_pack,
        )
        # run_meta.json on disk lists the quarantine file
        rm = json.loads((out_dir / "run_meta.json").read_text(
            encoding="utf-8"))
        assert "background_evidence_pack_quarantined.json" in rm["outputs"]
        assert rm["run_status"] == "validation_failed"
        assert rm["exit_code"] == 1
        assert rm["filter_validation_passed"] is False
        assert rm["quarantine_output"] == (
            "background_evidence_pack_quarantined.json"
        )

    def test_html_first_paint_shows_validation_failed_warning(self, tmp_path):
        """HTML rendered for a validation_failed run must show a red /
        warning banner and NOT display the dry-run badge."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        invalid_pack = _invalid_pack_with_high_confidence(bundle)
        filter_rep = bsx.run_evidence_filter_validation(
            evidence_pack=invalid_pack, bundle=bundle, schema=fs,
        )
        filter_rep["run_status"] = "validation_failed"
        filter_rep["generate_status"] = "succeeded"
        placeholder = {
            "placeholder": True, "stage": "evidence_filter",
            "source_bundle_ref": bundle.bundle_id,
            "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
            "reason": "validation failed; quarantined",
        }
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
            run_status="validation_failed",
            exit_code=1,
            filter_validation_passed=False,
            generate_status="succeeded",
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=placeholder,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=bsx.planned_checker_report(),
            run_meta=run_meta, include_diagnostic=False,
        )
        low = html.lower()
        # No dry-run badge when actually generating.
        assert "w1 dry-run (no llm call)" not in low, (
            "HTML must not show the dry-run badge on a generate run"
        )
        # validation_failed badge present.
        assert "validation_failed" in low or "validation failed" in low, (
            "HTML must explicitly say validation failed"
        )
        # Placeholder messaging must distinguish validation_failed from
        # dry-run (not "Filter LLM not called in dry-run").
        assert "filter llm not called in dry-run" not in low, (
            "HTML must not claim the LLM was not called when it actually "
            "was; this run is validation_failed"
        )

    def test_html_dry_run_badge_appears_when_actually_dry_run(self, tmp_path):
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        ph_pack = bsx.placeholder_evidence_pack(bundle)
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": False}, outputs=[],
            model="model_for_future_generate",
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=ph_pack,
            evidence_filter_report=bsx.planned_evidence_filter_report(),
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=bsx.planned_checker_report(),
            run_meta=run_meta, include_diagnostic=False,
        )
        low = html.lower()
        assert "dry-run" in low, "dry-run badge must appear on a dry-run"


class TestPlanVersionAndCliDescriptionMatchW2a:
    def test_plan_version_bumped_past_w1(self):
        # Anything other than the bare W1 marker is fine — w2a / w2a-b /
        # bsx_w2a_b / w2a-d etc. The point is that PLAN_VERSION must
        # not still say "bsx_w1" while the script implements the
        # 3-stage gateway + correction pass.
        assert bsx.PLAN_VERSION != "bsx_w1", (
            "PLAN_VERSION must move past bsx_w1 once W2a/W2a-b ship"
        )
        assert "w2" in bsx.PLAN_VERSION.lower(), bsx.PLAN_VERSION

    def test_plan_version_reflects_correction_wave(self):
        # W2a-d added the correction retry. The PLAN_VERSION should bump
        # past W2a-b so downstream tooling knows the contract changed.
        # Accept any marker after w2a_b: w2a_c / w2a_d / w2b / etc.
        assert bsx.PLAN_VERSION != "bsx_w2a_b", (
            "PLAN_VERSION must move past bsx_w2a_b once W2a-d "
            "(correction retry) ships"
        )


# ---------------------------------------------------------------------------
# 6-K. (RETIRED) W2a-f role/applies_to refinement was superseded by
# W2-simple BackgroundContinuityBrief (Codex consultation 2026-05-24).
# The W2a-f test class definitions below were removed during the
# transition; see TestW2SimpleBriefContract (6-L) for the active
# downstream-input contract tests.
# ---------------------------------------------------------------------------
class _W2aFRetired:
    def test_role_and_applies_to_enums_are_generic(self):
        """Generic enum constants — no scenario-specific tokens."""
        assert set(bsx.EVIDENCE_PACK_ROLE_ENUM) == {
            "global_identity", "world_default",
            "layout_feel", "state_variation",
            "avoid_default", "gap",
        }
        assert set(bsx.EVIDENCE_PACK_APPLIES_TO_ENUM) == {
            "master_background", "specific_state_only",
            "shot_specific", "avoid", "unknown",
        }

    def test_legacy_keys_are_listed_in_module_constant(self):
        assert "use_for" in bsx.EVIDENCE_PACK_LEGACY_KEYS
        assert "pass_down_hint" in bsx.EVIDENCE_PACK_LEGACY_KEYS
        assert "why_background_relevant" in bsx.EVIDENCE_PACK_LEGACY_KEYS

    def test_checker_rejects_pack_with_legacy_pass_down_hint(self):
        _, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        pack = _evidence_pack_basic(bundle)
        # Re-introduce the legacy key on an item.
        pack["evidence_items"][0]["pass_down_hint"] = "this is a hint"
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=schema,
        )
        assert not rep["passed"]
        assert any("forbidden_key" in f and "pass_down_hint" in f
                   for f in rep["failed_checks"])

    def test_checker_does_not_judge_role_semantics(self):
        """Even if `state_variation` content is mis-tagged as
        `global_identity`, the checker passes — deterministic code is
        not allowed to second-guess role semantics, only enum
        membership."""
        _, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        pack = _evidence_pack_basic(bundle)
        # All enums are valid; only the semantic mapping is questionable.
        pack["evidence_items"][0]["role"] = "global_identity"
        pack["evidence_items"][0]["applies_to"] = "master_background"
        pack["evidence_items"][0]["background_fact"] = (
            "a temporary marking observed during one scene"
        )
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=schema,
        )
        # Schema-shape PASS — semantic correctness is out of scope for
        # the deterministic checker.
        assert rep["passed"], rep["failed_checks"]

    def test_evidence_filter_system_prompt_forbids_image_commands_and_separates_master_state(self):
        spec = bsx.build_sample_fixture_l05_spec()
        sys_txt = bsx.build_prompt_evidence_filter_system(spec)
        low = sys_txt.lower()
        # Image-command ban surface.
        assert ("do not write image prompts" in low
                or "image prompt" in low and "not" in low), (
            "system prompt must forbid image-prompt directives"
        )
        # Master / state separation surface.
        assert "state_variation" in low
        assert "global_identity" in low
        # Generic vocabulary only — no sample-specific scene words.
        forbidden_sample_tokens = (
            "옥탑", "옥탑방", "rooftop", "옥상", "민숙",
            "L05", "수리영", "안방",
        )
        for tok in forbidden_sample_tokens:
            assert tok.lower() not in low, (
                f"system prompt must not embed sample-specific token "
                f"{tok!r}"
            )

    def test_evidence_filter_user_prompt_lists_role_and_applies_to_enums(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        text = bsx.build_prompt_evidence_filter_user(
            spec=spec, bundle=bundle, schema=schema,
        )
        low = text.lower()
        for tok in ("global_identity", "world_default", "layout_feel",
                     "state_variation", "avoid_default", "gap"):
            assert tok in low, f"user prompt must list role enum {tok!r}"
        for tok in ("master_background", "specific_state_only",
                     "shot_specific", "avoid", "unknown"):
            assert tok in low, (
                f"user prompt must list applies_to enum {tok!r}"
            )


class _W2aFRetiredHtmlDownstreamFirst:
    def _render(self, bundle, evidence_pack, run_status="succeeded",
                 generate_status="succeeded",
                 filter_validation_passed=True):
        spec = bsx.build_sample_fixture_l05_spec()
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        filter_rep = (
            bsx.run_evidence_filter_validation(
                evidence_pack=evidence_pack, bundle=bundle, schema=fs,
            ) if not evidence_pack.get("placeholder")
            else bsx.planned_evidence_filter_report()
        )
        filter_rep["run_status"] = run_status
        filter_rep["generate_status"] = generate_status
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
            run_status=run_status, exit_code=0,
            filter_validation_passed=filter_validation_passed,
            generate_status=generate_status,
            filter_failed_checks=[], initial_failed_checks=[],
        )
        return bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=evidence_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=bsx.planned_checker_report(),
            run_meta=run_meta, include_diagnostic=False,
        )

    def test_html_first_paint_shows_refined_pack_before_raw_sourcebundle(self):
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="coastal village at present day plus more",
            episode_text="ep body",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        pack = _evidence_pack_basic(bundle)
        # The fixture references "planning_doc:test"; rewrite source_ref
        # to match the real synthetic source_ref so quote containment
        # holds.
        planning_ref = next(
            s.source_ref for s in bundle.sources if s.kind == "planning_doc"
        )
        pack["evidence_items"][0]["source_ref"] = planning_ref
        pack["source_bundle_ref"] = bundle.bundle_id
        html = self._render(bundle, pack)
        low = html.lower()
        refined_idx = low.find("refined downstream evidence pack")
        # Use the unique §99 anchor to locate the raw-preview section.
        raw_section_idx = low.find(
            "show raw sourcebundle table"
        )
        assert refined_idx >= 0, (
            "HTML must contain the refined-pack section header"
        )
        assert raw_section_idx >= 0, (
            "HTML must still surface the raw SourceBundle preview "
            "(but inside a collapsed details element near the bottom)"
        )
        assert refined_idx < raw_section_idx, (
            "Refined pack section must appear BEFORE the raw "
            "SourceBundle preview"
        )

    def test_html_role_distribution_uses_generic_labels(self):
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="coastal village at present day plus more",
            episode_text="ep body",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        pack = _evidence_pack_basic(bundle)
        planning_ref = next(
            s.source_ref for s in bundle.sources if s.kind == "planning_doc"
        )
        pack["evidence_items"][0]["source_ref"] = planning_ref
        pack["source_bundle_ref"] = bundle.bundle_id
        html = self._render(bundle, pack)
        low = html.lower()
        assert "role distribution" in low
        # The role token chosen by the LLM (here: global_identity)
        # appears, but no sample-specific token from the bundle leaks
        # into role labels.
        assert "global_identity" in low

    def test_cli_description_mentions_evidence_filter_or_3_stage(self):
        ap = bsx.parse_args.__globals__["argparse"].ArgumentParser
        # Easier: re-call parse_args(["--help-noop"]) is tricky; just
        # inspect the parser by reconstructing it.
        # Build a fresh parser via parse_args to read description.
        # parse_args returns Namespace; the description is set in the
        # argparse.ArgumentParser ctor in parse_args(). We grab it by
        # introspecting source.
        src = SCRIPT_PATH.read_text(encoding="utf-8")
        # The description string is inside parse_args.
        assert "evidence filter" in src.lower() or "3-stage" in src.lower() or "background evidence pack" in src.lower(), (
            "CLI description should mention Evidence Filter / "
            "BackgroundEvidencePack / 3-stage flow somewhere"
        )


# ---------------------------------------------------------------------------
# 6-J. W2a-d Evidence Filter correction pass (Codex 권장 2026-05-24).
#
# When the initial Evidence Filter pack fails deterministic validation
# because some quotes are not present verbatim in their cited source,
# the script must call the LLM ONE more time with a correction prompt
# (failed_checks + the original SourceBundle, no truncation) and re-
# validate. Promote on success; quarantine on failure. Correction is
# triggered ONLY when quote_not_in_source is among the failed_checks;
# enum / cap / schema-shape failures are conservative fail-closed.
# ---------------------------------------------------------------------------
def _evidence_pack_with_one_bad_quote(bundle):
    """Pack whose only failure is one quote not present in the source.
    confidence_band / role / applies_to / slug / cap are all valid."""
    return {
        "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
        "source_bundle_ref": bundle.bundle_id,
        "evidence_items": [
            {
                "evidence_id": "ev_present_quote",
                "source_ref": "planning_doc:test",
                "quote": "coastal village at present day",
                "background_fact": "baseline place identity",
                "role": "global_identity",
                "applies_to": "master_background",
                "why_keep": "stable place anchor",
                "confidence_band": "trusted",
            },
            {
                "evidence_id": "ev_bad_quote",
                "source_ref": "planning_doc:test",
                "quote": "this paraphrase does not appear in the source",
                "background_fact": "fabricated layout feel",
                "role": "layout_feel",
                "applies_to": "master_background",
                "why_keep": "spatial scale anchor",
                "confidence_band": "plausible",
            },
        ],
        "coverage_notes": [],
        "rejected_or_irrelevant_summary": [],
    }


def _evidence_pack_with_enum_only_failure(bundle):
    """Pack whose failure is enum-only (no quote_not_in_source). Correction
    must NOT be triggered for this — fail-closed conservatively."""
    return {
        "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
        "source_bundle_ref": bundle.bundle_id,
        "evidence_items": [
            {
                "evidence_id": "ev_enum_bad",
                "source_ref": "planning_doc:test",
                "quote": "coastal village at present day",
                "background_fact": "baseline place identity",
                "role": "global_identity",
                "applies_to": "master_background",
                "why_keep": "stable place anchor",
                "confidence_band": "high",  # invalid enum
            },
        ],
        "coverage_notes": [],
        "rejected_or_irrelevant_summary": [],
    }


class TestCorrectionPromptBuilders:
    """Prompt builder signature + AST contract."""

    def test_correction_prompt_builders_exist_and_return_strings(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        invalid_pack = _evidence_pack_with_one_bad_quote(bundle)
        sys_txt = bsx.build_prompt_evidence_filter_correction_system(spec)
        usr_txt = bsx.build_prompt_evidence_filter_correction_user(
            spec=spec, bundle=bundle, schema=schema,
            invalid_pack=invalid_pack,
            failed_checks=[
                "evidence_filter:quote_not_in_source:1",
            ],
        )
        assert isinstance(sys_txt, str) and sys_txt.strip()
        assert isinstance(usr_txt, str) and usr_txt.strip()
        # User prompt must surface the rejected quote and source_ref so
        # the LLM can locate which item to fix.
        assert "this paraphrase does not appear in the source" in usr_txt
        assert "planning_doc:test" in usr_txt
        # System prompt must spell out the verbatim discipline.
        low = sys_txt.lower()
        assert "verbatim" in low
        assert "source" in low

    def test_correction_user_prompt_signature_takes_full_bundle_and_invalid_pack(self):
        sig = inspect.signature(
            bsx.build_prompt_evidence_filter_correction_user,
        )
        params = set(sig.parameters.keys())
        for required in ("bundle", "schema", "invalid_pack",
                          "failed_checks"):
            assert required in params, (
                f"correction user prompt builder must accept {required!r}; "
                f"params={params}"
            )

    def test_correction_user_prompt_does_not_truncate_source(self):
        """AST: correction prompt builder body must not slice any source
        body field. Same discipline as the main builders — no
        textwrap.shorten / .truncate / [:N] on source text."""
        fn = _function_def_by_name(
            "build_prompt_evidence_filter_correction_user",
        )
        assert fn is not None
        body_src = ast.unparse(fn)
        for pat in ("textwrap.shorten", ".truncate(",
                     "src.text[:", "source.text[:"):
            assert pat not in body_src, (
                f"correction user prompt builder must not truncate "
                f"source text; offending pattern: {pat!r}"
            )

    def test_correction_user_prompt_includes_full_source_text(self):
        """End-to-end: when called with a non-trivial bundle, every
        source.text is reproduced verbatim inside the prompt."""
        spec = bsx.build_sample_fixture_l05_spec()
        long_text = "X" * 5000 + " distinctive marker " + "Y" * 5000
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text=long_text, episode_text="short episode body",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        schema = bsx.build_background_evidence_pack_schema(bundle)
        invalid_pack = {
            "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
            "source_bundle_ref": bundle.bundle_id,
            "evidence_items": [], "coverage_notes": [],
            "rejected_or_irrelevant_summary": [],
        }
        usr = bsx.build_prompt_evidence_filter_correction_user(
            spec=spec, bundle=bundle, schema=schema,
            invalid_pack=invalid_pack, failed_checks=[],
        )
        assert long_text in usr, (
            "correction prompt must embed the full planning_doc body "
            "verbatim (no truncation)"
        )
        assert "short episode body" in usr


class TestCorrectionRoutingDecision:
    """route_evidence_filter_result must invoke correction iff
    quote_not_in_source is present AND a correction_call is provided."""

    def test_correction_invoked_only_for_quote_not_in_source(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        # Enum-only failure: correction must be skipped (fail-closed).
        enum_pack = _continuity_brief_with_enum_only_failure(bundle)
        called: list[str] = []
        def _correction(*, bundle, schema, model, invalid_pack,
                         failed_checks):
            called.append("invoked")
            return invalid_pack  # would only matter if invoked
        filter_result = {
            "status": "succeeded", "attempts": 1,
            "evidence_pack": enum_pack,
            "model": "gemini-3.5-flash", "error": "",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
            correction_call=_correction,
            model="gemini-3.5-flash",
        )
        assert routed["run_status"] == "validation_failed"
        assert routed["exit_code"] == 1
        assert called == [], (
            "correction must be skipped when failed_checks has no "
            "quote_not_in_source entries"
        )
        rep = routed["evidence_filter_report"]
        assert rep.get("correction_attempted") is False
        assert rep.get("correction_status") == "skipped"

    def test_correction_invoked_when_quote_not_in_source_and_promotes_corrected_pack(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        bad_pack = _continuity_brief_with_one_bad_quote(bundle)
        # Correction returns a fully valid pack.
        good_pack = _evidence_pack_basic(bundle)
        called: list[dict] = []
        def _correction(*, bundle, schema, model, invalid_pack,
                         failed_checks):
            called.append({"failed": list(failed_checks),
                           "invalid_rule_id": invalid_pack["must_stay_consistent"][0]["rule_id"]})
            return good_pack
        filter_result = {
            "status": "succeeded", "attempts": 1,
            "evidence_pack": bad_pack,
            "model": "gemini-3.5-flash", "error": "",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
            correction_call=_correction,
            model="gemini-3.5-flash",
        )
        assert called, "correction must be invoked"
        assert any(
            "quote_not_in_source" in fc for fc in called[0]["failed"]
        ), called[0]["failed"]
        assert routed["run_status"] == "succeeded"
        assert routed["exit_code"] == 0
        assert routed["quarantined_pack"] is None
        assert routed["downstream_pack"] is good_pack
        rep = routed["evidence_filter_report"]
        assert rep["correction_attempted"] is True
        assert rep["correction_status"] == "succeeded"
        assert rep["evidence_filter_attempts"] >= 2
        # initial_failed_checks preserved for audit.
        assert "quote_not_in_source" in (
            ";".join(rep.get("initial_failed_checks") or [])
        )

    def test_correction_quarantines_corrected_pack_when_validation_still_fails(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        bad_pack = _continuity_brief_with_one_bad_quote(bundle)
        # Correction returns ANOTHER invalid pack — same failure mode.
        worse_pack = _continuity_brief_with_one_bad_quote(bundle)
        worse_pack["must_stay_consistent"][0]["evidence_refs"][1][
            "quote"] = "yet another quote not present in source"
        def _correction(*, bundle, schema, model, invalid_pack,
                         failed_checks):
            return worse_pack
        filter_result = {
            "status": "succeeded", "attempts": 1,
            "evidence_pack": bad_pack,
            "model": "gemini-3.5-flash", "error": "",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
            correction_call=_correction,
            model="gemini-3.5-flash",
        )
        assert routed["run_status"] == "validation_failed"
        assert routed["exit_code"] == 1
        # Quarantine the CORRECTED pack (latest attempt).
        assert routed["quarantined_pack"] is worse_pack
        assert routed["downstream_pack"].get("placeholder") is True
        rep = routed["evidence_filter_report"]
        assert rep["correction_attempted"] is True
        assert rep["correction_status"] == "validation_failed"
        assert rep.get("correction_failed_checks"), (
            "correction_failed_checks must list the post-correction fails"
        )

    def test_correction_handles_llm_call_exception(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        bad_pack = _continuity_brief_with_one_bad_quote(bundle)
        def _correction(*, bundle, schema, model, invalid_pack,
                         failed_checks):
            raise RuntimeError("api timeout during correction")
        filter_result = {
            "status": "succeeded", "attempts": 1,
            "evidence_pack": bad_pack,
            "model": "gemini-3.5-flash", "error": "",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
            correction_call=_correction,
            model="gemini-3.5-flash",
        )
        assert routed["run_status"] == "validation_failed"
        assert routed["exit_code"] == 1
        rep = routed["evidence_filter_report"]
        assert rep["correction_attempted"] is True
        assert rep["correction_status"] == "failed"
        assert "api timeout" in (rep.get("correction_error") or "")
        # Original invalid pack remains quarantined.
        assert routed["quarantined_pack"] is bad_pack

    def test_correction_not_attempted_when_initial_generate_failed(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        called: list[str] = []
        def _correction(*, bundle, schema, model, invalid_pack,
                         failed_checks):
            called.append("nope")
            return invalid_pack
        filter_result = {
            "status": "failed", "attempts": 2,
            "evidence_pack": None,
            "model": "gemini-3.5-flash",
            "error": "api unreachable",
        }
        routed = bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
            correction_call=_correction,
            model="gemini-3.5-flash",
        )
        assert routed["run_status"] == "generate_failed"
        assert routed["exit_code"] == 1
        assert called == [], (
            "correction must not run when the initial LLM call itself "
            "failed"
        )

    def test_correction_runs_at_most_once(self):
        """Even if the correction pack still has quote_not_in_source,
        the script must NOT loop and call correction again."""
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        bad_pack = _continuity_brief_with_one_bad_quote(bundle)
        still_bad = _continuity_brief_with_one_bad_quote(bundle)
        call_count = {"n": 0}
        def _correction(*, bundle, schema, model, invalid_pack,
                         failed_checks):
            call_count["n"] += 1
            return still_bad
        filter_result = {
            "status": "succeeded", "attempts": 1,
            "evidence_pack": bad_pack,
            "model": "gemini-3.5-flash", "error": "",
        }
        bsx.route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle, schema=schema,
            correction_call=_correction,
            model="gemini-3.5-flash",
        )
        assert call_count["n"] == 1, (
            f"correction must be called exactly once, got {call_count['n']}"
        )


class TestCorrectionRunMetaAndHtml:
    """run_meta + HTML expose correction status to the user."""

    def test_build_run_meta_carries_correction_keys(self):
        rm = bsx.build_run_meta(
            run_id="rt", fixture_id="fx",
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=10, bundle_sha256="x" * 64,
            run_status="succeeded", exit_code=0,
            filter_validation_passed=True, generate_status="succeeded",
            correction_attempted=True,
            correction_status="succeeded",
            correction_model="gemini-3.5-flash",
            correction_failed_checks=[],
            evidence_filter_attempts=2,
        )
        assert rm["correction_attempted"] is True
        assert rm["correction_status"] == "succeeded"
        assert rm["correction_model"] == "gemini-3.5-flash"
        assert rm["correction_failed_checks"] == []
        assert rm["evidence_filter_attempts"] == 2

    def test_build_run_meta_defaults_correction_fields(self):
        rm = bsx.build_run_meta(
            run_id="rt", fixture_id="fx",
            args_dict={"generate": False}, outputs=[],
            model="m", char_budget_estimate=0, bundle_sha256="x" * 64,
        )
        assert rm["correction_attempted"] is False
        assert rm["correction_status"] == "not_attempted"
        assert rm["correction_model"] is None
        assert rm["correction_failed_checks"] == []
        assert rm["evidence_filter_attempts"] == 0

    def test_build_run_meta_filter_failed_checks_means_final_only(self):
        """W2a-e BLOCKING 1: when correction succeeded, the top-level
        `filter_failed_checks` must be [] (final state). The pre-
        correction failures live under a separate `initial_failed_checks`
        field so downstream code cannot misread a success as a failure.
        """
        rm = bsx.build_run_meta(
            run_id="rt", fixture_id="fx",
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=10, bundle_sha256="x" * 64,
            run_status="succeeded", exit_code=0,
            filter_validation_passed=True, generate_status="succeeded",
            filter_failed_checks=[],
            initial_failed_checks=[
                "evidence_filter:quote_not_in_source:1",
                "evidence_filter:quote_not_in_source:5",
            ],
            correction_attempted=True,
            correction_status="succeeded",
            correction_model="gemini-3.5-flash",
            correction_failed_checks=[],
            evidence_filter_attempts=2,
        )
        assert rm["filter_validation_passed"] is True
        assert rm["filter_failed_checks"] == [], (
            "after-correction success must have empty filter_failed_checks"
        )
        assert rm["initial_failed_checks"] == [
            "evidence_filter:quote_not_in_source:1",
            "evidence_filter:quote_not_in_source:5",
        ]
        assert rm["correction_failed_checks"] == []
        assert rm["correction_attempted"] is True

    def test_main_run_meta_after_correction_does_not_carry_initial_fails_in_filter_failed_checks(
            self, monkeypatch, tmp_path):
        """End-to-end of the generate path with stub LLMs: an initial
        pack that fails `quote_not_in_source`, then a correction call
        that returns a valid pack. main()-equivalent run_meta on disk
        must have filter_failed_checks=[] and initial_failed_checks
        populated."""
        spec = bsx.build_sample_fixture_l05_spec()

        class _FakeSession:
            def __enter__(self):
                return self
            def __exit__(self, *a):
                return False
        def _fake_session_local():
            return _FakeSession()
        import app.core.database as _db
        monkeypatch.setattr(_db, "SessionLocal", _fake_session_local)
        monkeypatch.setattr(bsx, "load_planning_doc",
                            lambda s, pid: ("present-day coastal village body",
                                            {"id": pid}))
        monkeypatch.setattr(bsx, "load_episode_fulltext",
                            lambda s, eid: ("episode body", {"id": eid}))
        monkeypatch.setattr(
            bsx, "load_selected_shots_for_location",
            lambda s, pid, eid, lid: [],
        )
        monkeypatch.setattr(bsx, "load_entity_catalog",
                            lambda s, pid: [])
        monkeypatch.setattr(bsx, "load_location_catalog",
                            lambda s, pid: [])

        # Initial LLM returns a brief whose common_place_identity has a
        # bad evidence_ref quote (correction trigger).
        def _initial(*, bundle, schema, model):
            er_good = {
                "source_ref": "planning_doc:" + spec.project_id,
                "quote": "present-day coastal village body",
            }
            er_bad = {
                "source_ref": "planning_doc:" + spec.project_id,
                "quote": "coastal mythology not in source",
            }
            return {
                "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
                "source_bundle_ref": bundle.bundle_id,
                "common_place_identity": {
                    "summary": "baseline",
                    "why_this_identity": "baseline",
                    "confidence_band": "trusted",
                    "evidence_refs": [er_good, er_bad],
                },
                "must_stay_consistent": [],
                "must_not_contradict": [],
                "allowed_creative_freedom": [],
                "state_change_rules": [],
                "shot_conflict_checks": [],
                "unknowns_left_to_art_direction": [],
                "coverage_notes": [],
                "rejected_or_irrelevant_summary": [],
            }
        # Correction LLM returns a fully valid brief.
        def _correction(*, bundle, schema, model, invalid_pack,
                         failed_checks):
            er = {
                "source_ref": "planning_doc:" + spec.project_id,
                "quote": "present-day coastal village body",
            }
            return {
                "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
                "source_bundle_ref": bundle.bundle_id,
                "common_place_identity": {
                    "summary": "baseline",
                    "why_this_identity": "baseline",
                    "confidence_band": "trusted",
                    "evidence_refs": [er],
                },
                "must_stay_consistent": [],
                "must_not_contradict": [],
                "allowed_creative_freedom": [],
                "state_change_rules": [],
                "shot_conflict_checks": [],
                "unknowns_left_to_art_direction": [],
                "coverage_notes": [],
                "rejected_or_irrelevant_summary": [],
            }
        monkeypatch.setattr(bsx, "_default_evidence_filter_llm_call",
                            _initial)
        monkeypatch.setattr(
            bsx, "_default_evidence_filter_correction_llm_call",
            _correction,
        )

        out_root = tmp_path / "runs"
        rc = bsx.main([
            "--generate", "--model", "gemini-3.5-flash",
            "--no-serve", "--output-root", str(out_root),
        ])
        assert rc == 0, (
            "main must return 0 when correction promotes a valid pack"
        )
        run_dirs = list(out_root.iterdir())
        assert len(run_dirs) == 1
        run_dir = run_dirs[0]
        rm = json.loads(
            (run_dir / "run_meta.json").read_text(encoding="utf-8"))
        # ★ The top-level filter_failed_checks must reflect the FINAL
        # state, not the initial state. Otherwise downstream code reading
        # run_meta would misclassify this success as a failure.
        assert rm["filter_validation_passed"] is True
        assert rm["filter_failed_checks"] == [], (
            f"after-correction success must NOT carry initial fails in "
            f"run_meta.filter_failed_checks; got {rm['filter_failed_checks']}"
        )
        # initial_failed_checks must surface the pre-correction failures
        # so audits can trace what triggered the correction.
        assert any(
            "quote_not_in_source" in f
            for f in (rm.get("initial_failed_checks") or [])
        ), rm.get("initial_failed_checks")
        assert rm["correction_attempted"] is True
        assert rm["correction_status"] == "succeeded"

    def test_main_run_meta_after_correction_failed_marks_filter_failed_checks_to_latest(
            self, monkeypatch, tmp_path):
        """End-to-end: when correction is invoked but still fails, the
        top-level filter_failed_checks must reflect the LATEST
        (post-correction) failures, and initial_failed_checks holds the
        pre-correction failures."""
        spec = bsx.build_sample_fixture_l05_spec()
        class _FakeSession:
            def __enter__(self):
                return self
            def __exit__(self, *a):
                return False
        def _fake_session_local():
            return _FakeSession()
        import app.core.database as _db
        monkeypatch.setattr(_db, "SessionLocal", _fake_session_local)
        monkeypatch.setattr(bsx, "load_planning_doc",
                            lambda s, pid: ("present-day coastal village body",
                                            {"id": pid}))
        monkeypatch.setattr(bsx, "load_episode_fulltext",
                            lambda s, eid: ("episode body", {"id": eid}))
        monkeypatch.setattr(
            bsx, "load_selected_shots_for_location",
            lambda s, pid, eid, lid: [],
        )
        monkeypatch.setattr(bsx, "load_entity_catalog",
                            lambda s, pid: [])
        monkeypatch.setattr(bsx, "load_location_catalog",
                            lambda s, pid: [])

        def _bad_initial(*, bundle, schema, model):
            return {
                "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
                "source_bundle_ref": bundle.bundle_id,
                "common_place_identity": {
                    "summary": "baseline",
                    "why_this_identity": "baseline",
                    "confidence_band": "trusted",
                    "evidence_refs": [{
                        "source_ref": "planning_doc:" + spec.project_id,
                        "quote": "initial paraphrase not in source",
                    }],
                },
                "must_stay_consistent": [],
                "must_not_contradict": [],
                "allowed_creative_freedom": [],
                "state_change_rules": [],
                "shot_conflict_checks": [],
                "unknowns_left_to_art_direction": [],
                "coverage_notes": [],
                "rejected_or_irrelevant_summary": [],
            }
        def _still_bad_correction(*, bundle, schema, model,
                                   invalid_pack, failed_checks):
            return {
                "schema_version": bsx.EVIDENCE_PACK_SCHEMA_VERSION,
                "source_bundle_ref": bundle.bundle_id,
                "common_place_identity": {
                    "summary": "baseline v2",
                    "why_this_identity": "baseline",
                    "confidence_band": "trusted",
                    "evidence_refs": [{
                        "source_ref": "planning_doc:" + spec.project_id,
                        "quote": "still wrong after correction",
                    }],
                },
                "must_stay_consistent": [],
                "must_not_contradict": [],
                "allowed_creative_freedom": [],
                "state_change_rules": [],
                "shot_conflict_checks": [],
                "unknowns_left_to_art_direction": [],
                "coverage_notes": [],
                "rejected_or_irrelevant_summary": [],
            }
        monkeypatch.setattr(bsx, "_default_evidence_filter_llm_call",
                            _bad_initial)
        monkeypatch.setattr(
            bsx, "_default_evidence_filter_correction_llm_call",
            _still_bad_correction,
        )

        out_root = tmp_path / "runs"
        rc = bsx.main([
            "--generate", "--model", "gemini-3.5-flash",
            "--no-serve", "--output-root", str(out_root),
        ])
        assert rc != 0
        run_dirs = list(out_root.iterdir())
        assert len(run_dirs) == 1
        rm = json.loads(
            (run_dirs[0] / "run_meta.json").read_text(encoding="utf-8"))
        assert rm["filter_validation_passed"] is False
        # Top-level reflects the LATEST failures (the correction round).
        assert any(
            "quote_not_in_source" in f
            for f in rm["filter_failed_checks"]
        )
        # Pre-correction failures preserved separately.
        assert any(
            "quote_not_in_source" in f
            for f in (rm.get("initial_failed_checks") or [])
        )
        assert rm["correction_attempted"] is True
        assert rm["correction_status"] == "validation_failed"

    def test_html_announces_initial_failures_corrected(self, tmp_path):
        """When a run succeeded after correction, the HTML should make
        the audit trail visible: 'initial ... corrected: N' style."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        good_pack = _evidence_pack_basic(bundle)
        filter_rep = bsx.run_evidence_filter_validation(
            evidence_pack=good_pack, bundle=bundle, schema=fs,
        )
        filter_rep["run_status"] = "succeeded"
        filter_rep["generate_status"] = "succeeded"
        filter_rep["correction_attempted"] = True
        filter_rep["correction_status"] = "succeeded"
        filter_rep["initial_failed_checks"] = [
            "evidence_filter:quote_not_in_source:1",
            "evidence_filter:quote_not_in_source:5",
        ]
        filter_rep["evidence_filter_attempts"] = 2
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
            run_status="succeeded", exit_code=0,
            filter_validation_passed=True, generate_status="succeeded",
            filter_failed_checks=[],
            initial_failed_checks=[
                "evidence_filter:quote_not_in_source:1",
                "evidence_filter:quote_not_in_source:5",
            ],
            correction_attempted=True,
            correction_status="succeeded",
            correction_model="gemini-3.5-flash",
            correction_failed_checks=[],
            evidence_filter_attempts=2,
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=good_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=bsx.planned_checker_report(),
            run_meta=run_meta, include_diagnostic=False,
        )
        low = html.lower()
        # Some signal that initial failures existed and were corrected.
        assert "initial" in low and "corrected" in low, (
            "HTML should expose the initial-failures audit trail when "
            "correction succeeded (e.g. 'initial quote failures "
            "corrected: 2')"
        )

    def test_html_badge_announces_correction_when_succeeded_after_correction(
            self, tmp_path):
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="p", episode_text="e",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        good_pack = _evidence_pack_basic(bundle)
        # Use a passing report.
        filter_rep = bsx.run_evidence_filter_validation(
            evidence_pack=good_pack, bundle=bundle, schema=fs,
        )
        filter_rep["run_status"] = "succeeded"
        filter_rep["generate_status"] = "succeeded"
        filter_rep["correction_attempted"] = True
        filter_rep["correction_status"] = "succeeded"
        filter_rep["evidence_filter_attempts"] = 2
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
            run_status="succeeded", exit_code=0,
            filter_validation_passed=True,
            generate_status="succeeded",
            correction_attempted=True,
            correction_status="succeeded",
            correction_model="gemini-3.5-flash",
            correction_failed_checks=[],
            evidence_filter_attempts=2,
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=good_pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=bsx.planned_checker_report(),
            run_meta=run_meta, include_diagnostic=False,
        )
        low = html.lower()
        assert "after correction" in low, (
            "HTML must announce the correction pass when run_status="
            "succeeded with correction_attempted=True"
        )


# ---------------------------------------------------------------------------
# 6-L. W2-simple BackgroundContinuityBrief contract (Codex consultation
# APPROVED with revisions, 2026-05-24). Tests use only synthetic
# generic data — never sample-specific terms.
# ---------------------------------------------------------------------------
class TestW2SimpleBriefContract:
    def test_plan_version_marks_w2_simple(self):
        assert bsx.PLAN_VERSION == "bsx_w2_simple"

    def test_evidence_filter_system_prompt_uses_constraint_language(self):
        spec = bsx.build_sample_fixture_l05_spec()
        sys_txt = bsx.build_prompt_evidence_filter_system(spec).lower()
        # constraint vocabulary surface
        assert "constraint" in sys_txt
        assert "do not write image" in sys_txt or "image command" in sys_txt
        # mention of the 9 brief sections + shot_ref exact-equality rule
        for token in (
            "common_place_identity", "must_stay_consistent",
            "must_not_contradict", "allowed_creative_freedom",
            "state_change_rules", "shot_conflict_checks",
            "unknowns_left_to_art_direction",
            "coverage_notes", "rejected_or_irrelevant_summary",
        ):
            assert token in sys_txt, f"system prompt missing {token!r}"
        # shot_ref must be tied to selected_shots in the SourceBundle
        # (exact equality). The prompt should communicate that obligation
        # using either "exactly" or "exact equality"-style language.
        assert ("must be exactly" in sys_txt
                or "exact equality" in sys_txt
                or "exact-equal" in sys_txt
                or "exact equal" in sys_txt), sys_txt
        assert "selected_shot" in sys_txt
        # caps surface
        assert "≤ 24" in sys_txt or "24" in sys_txt
        # confidence belongs on the rule, not on the evidence_ref
        assert ("do not add `confidence_band` to evidence_refs"
                in sys_txt) or ("confidence sits on the rule" in sys_txt)
        # No sample-specific vocabulary leaks.
        for forbidden in (
            "옥탑", "옥탑방", "rooftop", "옥상", "민숙",
            "수리영", "안방",
        ):
            assert forbidden.lower() not in sys_txt, (
                f"system prompt must not embed sample-specific token "
                f"{forbidden!r}"
            )

    def test_evidence_filter_user_prompt_lists_allowed_basis_enum(self):
        spec, bundle = _minimal_synthetic_bundle()
        schema = bsx.build_background_evidence_pack_schema(bundle)
        text = bsx.build_prompt_evidence_filter_user(
            spec=spec, bundle=bundle, schema=schema,
        ).lower()
        for tok in (
            "not_specified", "weakly_constrained", "art_direction_choice",
        ):
            assert tok in text, (
                f"user prompt must list allowed_creative_freedom basis "
                f"enum value {tok!r}"
            )

    def test_brief_html_first_paint_shows_brief_summary_before_raw_source(
            self):
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="coastal village at present day plus body",
            episode_text="ep body",
            selected_shots=[], entity_catalog=[], location_catalog=[],
        )
        pack = _evidence_pack_basic(bundle)
        # Use the real planning source_ref so quote containment holds,
        # and drop the placeholder shot section.
        planning_ref = next(
            s.source_ref for s in bundle.sources if s.kind == "planning_doc"
        )
        pack["source_bundle_ref"] = bundle.bundle_id
        for section in (
            "must_stay_consistent", "must_not_contradict",
            "state_change_rules",
        ):
            for it in pack[section]:
                for er in it.get("evidence_refs") or []:
                    er["source_ref"] = planning_ref
        for er in pack["common_place_identity"]["evidence_refs"]:
            er["source_ref"] = planning_ref
        pack["shot_conflict_checks"] = []
        ws = bsx.build_world_brief_schema(bundle)
        ms = bsx.build_minimal_spatial_brief_schema(bundle)
        fs = bsx.build_background_evidence_pack_schema(bundle)
        filter_rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=fs,
        )
        filter_rep["run_status"] = "succeeded"
        filter_rep["generate_status"] = "succeeded"
        run_meta = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": True}, outputs=[],
            model="gemini-3.5-flash",
            char_budget_estimate=bundle.char_budget_estimate,
            bundle_sha256="x" * 64,
            run_status="succeeded", exit_code=0,
            filter_validation_passed=True,
            generate_status="succeeded",
        )
        html = bsx.render_html(
            bundle=bundle, world_schema=ws, minimal_schema=ms,
            evidence_pack_schema=fs, evidence_pack=pack,
            evidence_filter_report=filter_rep,
            prompt_evidence_filter_system="filter sys",
            prompt_evidence_filter_user="filter usr",
            prompt_world_system="sys w", prompt_world_user="usr w",
            prompt_spatial_system="sys s", prompt_spatial_user="usr s",
            validation_report=bsx.planned_checker_report(),
            run_meta=run_meta, include_diagnostic=False,
        )
        low = html.lower()
        brief_idx = low.find("backgroundcontinuitybrief — downstream input")
        cpi_idx = low.find("common_place_identity")
        raw_idx = low.find("show raw sourcebundle table")
        assert brief_idx >= 0
        assert cpi_idx >= 0
        assert raw_idx >= 0
        assert brief_idx < raw_idx, (
            "BackgroundContinuityBrief downstream summary must appear "
            "BEFORE the raw SourceBundle preview"
        )
        assert cpi_idx < raw_idx, (
            "common_place_identity section must appear BEFORE the raw "
            "SourceBundle preview"
        )
        # Brief identity summary must be visible.
        assert "shared baseline" in low

    def test_shot_conflict_check_requires_matching_shot_evidence(self):
        """W2-simple-b BLOCKING 1: per-shot conflict checks need
        at least one evidence_ref whose source_ref equals the rule's
        shot_ref. Episode-only refs are not sufficient grounding."""
        spec = bsx.build_sample_fixture_l05_spec()
        bundle = bsx.build_source_bundle(
            spec=spec, run_id="rt",
            planning_text="coastal village at present day",
            episode_text="distinctive episode body marker",
            selected_shots=[{
                "still_id": "abc-123",
                "scene_index": 1, "shot_index": 1,
                "shot_description": "shot body distinctive marker",
                "scene_summary": "scene summary",
                "visible_entities_json": "[]",
                "loc_short_ids_visible": [],
                "is_selected": True, "status": None,
            }],
            entity_catalog=[], location_catalog=[],
        )
        pack = _evidence_pack_basic(bundle)
        # Rewrite generic evidence_refs to use the real planning_doc
        # source_ref so the rest of the brief still passes.
        planning_ref = next(
            s.source_ref for s in bundle.sources if s.kind == "planning_doc"
        )
        episode_ref = next(
            s.source_ref for s in bundle.sources if s.kind == "episode_fulltext"
        )
        for section in (
            "must_stay_consistent", "must_not_contradict",
            "state_change_rules",
        ):
            for it in pack[section]:
                for er in it.get("evidence_refs") or []:
                    er["source_ref"] = planning_ref
                    er["quote"] = "coastal village at present day"
        for er in pack["common_place_identity"]["evidence_refs"]:
            er["source_ref"] = planning_ref
            er["quote"] = "coastal village at present day"
        pack["source_bundle_ref"] = bundle.bundle_id
        # shot_conflict_check with the right shot_ref but evidence
        # only from the episode source — this MUST fail W2-simple-b.
        real_shot_ref = next(
            s.source_ref for s in bundle.sources if s.kind == "selected_shot"
        )
        pack["shot_conflict_checks"] = [{
            "shot_ref": real_shot_ref,
            "must_support": ["x"],
            "avoid_contradiction": ["y"],
            "confidence_band": "plausible",
            "evidence_refs": [{
                "source_ref": episode_ref,
                "quote": "distinctive episode body marker",
            }],
        }]
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep_fail = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert not rep_fail["passed"]
        assert any(
            "shot_conflict_check_missing_matching_shot_evidence" in f
            for f in rep_fail["failed_checks"]
        ), rep_fail["failed_checks"]
        # Add a matching per-shot evidence_ref → PASS.
        pack["shot_conflict_checks"][0]["evidence_refs"].append({
            "source_ref": real_shot_ref,
            "quote": "shot body distinctive marker",
        })
        rep_ok = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert rep_ok["passed"], rep_ok["failed_checks"]

    def test_run_meta_stages_advertise_background_continuity_brief(self):
        spec = bsx.build_sample_fixture_l05_spec()
        rm = bsx.build_run_meta(
            run_id="rt", fixture_id=spec.fixture_id,
            args_dict={"generate": False}, outputs=[],
            model="m", char_budget_estimate=0, bundle_sha256="x" * 64,
        )
        assert rm["stages"][0] == "background_continuity_brief", (
            "Run meta must advertise the W2-simple downstream input as "
            "the first stage"
        )

    def test_checker_does_not_judge_rule_semantics(self):
        """Even if a rule statement is semantically weak / wrong, the
        checker passes — deterministic code only checks shape /
        provenance / caps / enums."""
        _, bundle = _minimal_synthetic_bundle()
        pack = _evidence_pack_basic(bundle)
        pack["shot_conflict_checks"] = []
        # Semantically odd but schema-valid rule.
        pack["must_stay_consistent"][0]["statement"] = (
            "this constraint is overly specific and probably wrong, "
            "but it still parses."
        )
        ws = bsx.build_background_evidence_pack_schema(bundle)
        rep = bsx.run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=ws,
        )
        assert rep["passed"], rep["failed_checks"]