#!/usr/bin/env python3
"""Background semantic extractor experiment — W2a-d (2026-05-24).

scripts_output/background_semantic_extractor_experiment/<run_id>/ — 18
files when the Evidence Filter pack validates cleanly, or 19 files
(adding background_evidence_pack_quarantined.json) when validation
fails and the original pack is quarantined.

W2a (Evidence Filter gateway) + W2a-b (fail-closed routing + prompt
enum discipline + plan SOT sync) + W2a-c (run_meta/HTML status honesty
+ PLAN_VERSION bump) + W2a-d (LLM correction retry for
quote_not_in_source failures; deterministic code never paraphrases
quotes) + W2a-e (run_meta.filter_failed_checks reflects FINAL state,
separate initial_failed_checks). LLM/API access only when --generate
is set; DB read-only, production code 수정 0.

Pipeline (3-stage):
  Stage Gateway — BackgroundEvidencePack: LLM reads the SourceBundle
    ONCE and emits a compact filtered pack. Caps enforced by the
    deterministic checker; over-cap is FAIL (original verbatim, no
    auto-truncate).
  Stage A — BackgroundWorldBrief: world grounding 압축/필터링.
    plot/lore 차단. Receives ONLY the filtered pack — never the raw
    SourceBundle.
  Stage B — MinimalSpatialBrief : Stage A 의 short world hints +
    filtered pack 만 참조. Receives ONLY the filtered pack and Stage A
    hints — never the raw SourceBundle.

★★★ Standing rules ([[feedback-no-literal-substring-meaning]] /
[[feedback-no-scenario-specific-coding]]):
  - 패턴 인지 / regex / 글자 substring 매칭 / term boundary / 조사 / 글자 단위
    의미 구분 금지. literal lexicon matcher 0 (어떤 형태도).
  - `re` 모듈 import 0 (plan §6-A-1). slug 검증도 char loop.
  - sample fixture (L05/옥탑방/...) 는 SAMPLE_FIXTURE_* + SampleFixtureSpec
    안에만. generic engine 함수 body 에 0.
  - production code 수정 0.

Imports:
  - app.core.database.SessionLocal (read-only)
  - app.models.project: Episode, EntityCanon, SceneStill (read-only)
  - app.models.catalog: ProjectRegistry (read-only)
  - stdlib only — `re` 금지.

Output 18 files (run dir, +1 quarantine when validation_failed):
  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 (placeholder unless --generate),
  world_brief.json (placeholder unless --generate, W2b),
  minimal_spatial_brief.json (placeholder unless --generate, W2b),
  filter_validation_report.json (planned unless --generate),
  validation_report.json (planned unless --generate),
  run_meta.json,
  index.html,
  background_evidence_pack_quarantined.json (only when
    run_status=="validation_failed").
"""
from __future__ import annotations

import argparse
import hashlib
import html as html_lib
import json
import os
import sys
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional

# repo root + env --------------------------------------------------------------
_REPO_ROOT = Path(__file__).resolve().parents[2]
_BACKEND_ROOT = _REPO_ROOT / "backend"
if str(_BACKEND_ROOT) not in sys.path:
    sys.path.insert(0, str(_BACKEND_ROOT))


def _load_backend_env() -> None:
    """Load backend/.env so SessionLocal can connect to PostgreSQL."""
    env_path = _BACKEND_ROOT / ".env"
    if not env_path.exists():
        return
    for raw in env_path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip()
        if (value.startswith('"') and value.endswith('"')) or (
            value.startswith("'") and value.endswith("'")
        ):
            value = value[1:-1]
        if key and key not in os.environ:
            os.environ[key] = value


_load_backend_env()


# ============================================================================
# Plan metadata + caps
# ============================================================================
PLAN_VERSION = "bsx_w2_simple"
STAGE_A_SCHEMA_VERSION = "bsx_world_w0g"
STAGE_B_SCHEMA_VERSION = "bsx_minimal_w0g"
MODEL_PLACEHOLDER = "model_for_future_generate"

CONFIDENCE_BANDS: tuple[str, ...] = (
    "trusted", "plausible", "weak", "unknown",
)

BASE_PLATE_ROLES: tuple[str, ...] = (
    "identity_anchor", "optional", "avoid",
)

# Topology fields that must NOT appear in either schema output (plan §3 #8).
TOPOLOGY_FIELDS_FORBIDDEN: tuple[str, ...] = (
    "parent_place_group_id",
    "set_topology_id",
    "geometry_change_kind",
    "bound_unit_need_ids",
    "structural_version_id",
    "rollup_id",
    "decision_status",
)

# Caps (plan §3 #9 minimality_caps). cap = inclusive upper bound; equal-to-cap
# is PASS, over-cap is FAIL.
STAGE_A_FIELD_CHAR_CAPS: dict[str, int] = {
    "era_and_time_period": 160,
    "geographic_cultural_grounding": 200,
    "technology_and_material_baseline": 200,
    "social_economic_visual_tone": 200,
    "genre_mood_constraints": 220,
}
STAGE_A_BG_DO_NOT_ASSUME_MAX = 5
STAGE_A_BG_DO_NOT_ASSUME_ITEM_CHARS = 160
STAGE_A_IRRELEVANT_MAX = 8
STAGE_A_IRRELEVANT_ITEM_CHARS = 200

STAGE_B_PLACE_LABEL_CHARS = 80
STAGE_B_PLACE_ONE_SENTENCE_CHARS = 240
STAGE_B_WORLD_HINTS_MAX = 5
STAGE_B_WORLD_HINTS_ITEM_CHARS = 120
STAGE_B_CONTINUITY_GROUPS_MAX = 3
STAGE_B_CONTINUITY_GROUP_LABEL_CHARS = 80
STAGE_B_CONTINUITY_GROUP_LIST_MAX = 5
STAGE_B_CONTINUITY_GROUP_ITEM_CHARS = 80
STAGE_B_SPATIAL_RELATIONS_MAX = 5
STAGE_B_SPATIAL_RELATION_DESC_CHARS = 180
STAGE_B_VISUAL_ANCHORS_MAX = 5
STAGE_B_VISUAL_ANCHOR_DESC_CHARS = 180
STAGE_B_STATE_VARIATIONS_MAX = 5
STAGE_B_STATE_VARIATION_DESC_CHARS = 180
STAGE_B_STATE_VARIATION_LIST_MAX = 5
STAGE_B_STATE_VARIATION_ITEM_CHARS = 120
STAGE_B_GENERATION_NOTES_BASE_PLATE_BRIEF_CHARS = 900
STAGE_B_GENERATION_NOTES_ARRAY_MAX = 5
STAGE_B_GENERATION_NOTES_ARRAY_ITEM_CHARS = 160
STAGE_B_SLUG_CHARS = 60

# ----------------------------------------------------------------------------
# Evidence Filter (W2a — Codex 2026-05-24).
#
# Stage Gateway: LLM reads the full SourceBundle ONCE and emits a compact
# BackgroundEvidencePack. Downstream Stage A / Stage B prompts receive ONLY
# the pack, never the raw bundle.
# ----------------------------------------------------------------------------
EVIDENCE_PACK_SCHEMA_VERSION = "bsx_continuity_brief_w2_simple"

# W2-simple BackgroundContinuityBrief: the LLM gateway produces a
# constraint brief (not a spatial decomposition). The downstream world /
# spatial stages must NOT contradict it; everything else is left to art
# direction.
CONTINUITY_BRIEF_RULE_ID_CHARS = 60
CONTINUITY_BRIEF_STATEMENT_CHARS = 200
CONTINUITY_BRIEF_WHY_CHARS = 180
CONTINUITY_BRIEF_LIST_MAX = 8
CONTINUITY_BRIEF_LIST_ITEM_CHARS = 160
CONTINUITY_BRIEF_SHOT_CHECKS_MAX = 24
CONTINUITY_BRIEF_IDENTITY_SUMMARY_CHARS = 240
CONTINUITY_BRIEF_IDENTITY_WHY_CHARS = 180

CONTINUITY_BRIEF_ALLOWED_BASIS_ENUM: tuple[str, ...] = (
    "not_specified",
    "weakly_constrained",
    "art_direction_choice",
)

EVIDENCE_PACK_COVERAGE_NOTES_MAX = 8
EVIDENCE_PACK_COVERAGE_NOTE_CHARS = 200
EVIDENCE_PACK_COVERAGE_CONSEQUENCE_CHARS = 200

EVIDENCE_PACK_REJECTED_MAX = 8
EVIDENCE_PACK_REJECTED_NOTE_CHARS = 200
EVIDENCE_PACK_REJECTED_WHY_CHARS = 240

# Legacy item-level keys forbidden in the W2-simple ContinuityBrief
# (these were the W2a..W2a-f shapes; the brief restructures the pack
# around rules rather than per-item tags).
EVIDENCE_PACK_LEGACY_KEYS: tuple[str, ...] = (
    "use_for",
    "pass_down_hint",
    "why_background_relevant",
    "role",
    "applies_to",
    "background_fact",
    "why_keep",
    "evidence_items",
)

EVIDENCE_PACK_FORBIDDEN_INVENTORY_KEYS: tuple[str, ...] = (
    "count",
    "door_count",
    "window_count",
    "furniture_count",
    "object_count",
    "inventory",
)

EVIDENCE_PACK_DEFAULT_GENERATE_MODEL = "gemini-3.5-flash"

# ============================================================================
# SAMPLE FIXTURE — generic methodology 검증용 표본 only. Generic engine
# 함수에 직접 박지 말 것.
# ============================================================================
SAMPLE_FIXTURE_PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19"
SAMPLE_FIXTURE_EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c"
SAMPLE_FIXTURE_L05_CANON_ID = "3afbc7a8-b919-4431-a691-0a99057a26ca"
SAMPLE_FIXTURE_L05_SHORT_ID = "L05"
SAMPLE_FIXTURE_SOURCE_RUN = Path(
    "scripts_output/rooftop_source_grounding/codex_entry_sanity_gemini_ok"
)
SAMPLE_FIXTURE_SOURCE_BIBLE_FILENAME = "gemini_rooftop_bible.json"

SAMPLE_FIXTURE_L05_MODEL_CONTEXT_CHAR_LIMIT = 300_000

DEFAULT_OUTPUT_DIR = Path(
    "scripts_output/background_semantic_extractor_experiment"
)


# ============================================================================
# Data classes
# ============================================================================
@dataclass
class SampleFixtureSpec:
    """All sample-specific data for a single fixture (e.g. L05 rooftop)."""
    fixture_id: str
    project_id: str
    episode_id: str
    canon_id: str
    location_short_id: str
    source_run_path: Path
    source_bible_filename: str
    default_model: str = MODEL_PLACEHOLDER
    model_context_char_limit: int = SAMPLE_FIXTURE_L05_MODEL_CONTEXT_CHAR_LIMIT


@dataclass
class SourceItem:
    source_ref: str
    kind: str
    text: str
    sha256: str
    char_count: int
    extras: dict[str, Any] = field(default_factory=dict)


@dataclass
class SourceBundle:
    bundle_id: str
    fixture_id: str
    project_id: str
    episode_id: str
    location_short_id: str
    sources: list[SourceItem] = field(default_factory=list)
    missing_inputs: list[dict] = field(default_factory=list)

    @property
    def char_budget_estimate(self) -> int:
        return sum(s.char_count for s in self.sources) + 4_000


# ============================================================================
# Sample fixture builder
# ============================================================================
def build_sample_fixture_l05_spec() -> SampleFixtureSpec:
    return SampleFixtureSpec(
        fixture_id="l05_rooftop_interior",
        project_id=SAMPLE_FIXTURE_PROJECT_ID,
        episode_id=SAMPLE_FIXTURE_EPISODE_ID,
        canon_id=SAMPLE_FIXTURE_L05_CANON_ID,
        location_short_id=SAMPLE_FIXTURE_L05_SHORT_ID,
        source_run_path=SAMPLE_FIXTURE_SOURCE_RUN,
        source_bible_filename=SAMPLE_FIXTURE_SOURCE_BIBLE_FILENAME,
    )


# ============================================================================
# Hashing / normalization / slug validation — no regex, char loop only.
# ============================================================================
def _sha256_hex(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _coerce_source_text(text: Optional[str]) -> str:
    """W1c: verbatim source preservation. None → "", otherwise return the
    original string unchanged (no strip, no NFC normalize, no truncation).
    """
    return text if isinstance(text, str) else ""


def _slug_is_valid(name: str) -> bool:
    """ASCII alnum + `_`/`-` only — char loop, no regex (plan §6-A-1)."""
    if not name:
        return False
    for ch in name:
        if not (ch.isalnum() or ch in "_-"):
            return False
    return True


# ============================================================================
# DB loaders — read-only. selected_shots = JSON parse + short_id exact
# equality (plan §1 / §6-B-6, no substring matching).
# ============================================================================
def load_planning_doc(session, project_id: str) -> tuple[str, dict]:
    from app.models.catalog import ProjectRegistry
    proj = session.query(ProjectRegistry).filter(
        ProjectRegistry.id == project_id,
    ).one_or_none()
    if proj is None:
        return "", {"missing": True, "id": project_id}
    # W1c: verbatim — no .strip(), preserve original text including
    # leading/trailing whitespace.
    text = _coerce_source_text(proj.planning_doc_text)
    return text, {
        "id": proj.id, "name": proj.name, "length_chars": len(text),
    }


def load_episode_fulltext(session, episode_id: str) -> tuple[str, dict]:
    from app.models.project import Episode
    ep = session.query(Episode).filter(Episode.id == episode_id).one_or_none()
    if ep is None:
        return "", {"missing": True, "id": episode_id}
    # W1c: verbatim — no .strip().
    text = _coerce_source_text(ep.fulltext)
    return text, {
        "id": ep.id,
        "episode_number": ep.episode_number,
        "title": ep.title,
        "length_chars": len(text),
    }


def load_selected_shots_for_location(session, project_id: str,
                                      episode_id: str,
                                      location_short_id: str) -> list[dict]:
    """plan §1 / BLOCKING 2: JSON parse visible_entities_json, include shot
    iff some entity dict has `short_id == location_short_id` (exact equality).
    NO substring matching on text.
    """
    from app.models.project import SceneStill
    rows = session.query(SceneStill).filter(
        SceneStill.episode_id == episode_id,
        SceneStill.project_id == project_id,
        SceneStill.is_selected.is_(True),
        SceneStill.still_index >= 0,
    ).order_by(
        SceneStill.scene_index, SceneStill.shot_index,
        SceneStill.still_index, SceneStill.id,
    ).all()
    out: list[dict] = []
    for r in rows:
        status = getattr(r, "status", None)
        if status == "stale":
            continue
        ve_raw = getattr(r, "visible_entities_json", None) or "[]"
        try:
            ve_parsed = (
                json.loads(ve_raw) if isinstance(ve_raw, str) else ve_raw
            )
        except (TypeError, ValueError):
            continue
        if not isinstance(ve_parsed, list):
            continue
        location_match = False
        loc_short_ids_visible: list[str] = []
        for elem in ve_parsed:
            if not isinstance(elem, dict):
                continue
            elem_short = elem.get("short_id")
            if isinstance(elem_short, str):
                loc_short_ids_visible.append(elem_short)
                if elem_short == location_short_id:
                    location_match = True
        if not location_match:
            continue
        # W1c: verbatim — no .strip() on shot_description / scene_summary.
        out.append({
            "still_id": r.id,
            "scene_index": getattr(r, "scene_index", None),
            "shot_index": getattr(r, "shot_index", None),
            "shot_description": _coerce_source_text(
                getattr(r, "shot_description", "")
            ),
            "scene_summary": _coerce_source_text(
                getattr(r, "scene_summary", "")
            ),
            "visible_entities_json": ve_raw,
            "loc_short_ids_visible": sorted(set(loc_short_ids_visible)),
            "is_selected": True,
            "status": status,
        })
    return out


def load_entity_catalog(session, project_id: str) -> list[dict]:
    from app.models.project import EntityCanon
    rows = session.query(EntityCanon).filter(
        EntityCanon.project_id == project_id,
        EntityCanon.entity_type.in_(["character", "prop", "outlook"]),
    ).order_by(
        EntityCanon.entity_type, EntityCanon.short_id, EntityCanon.id,
    ).all()
    out: list[dict] = []
    for r in rows:
        out.append({
            "canon_id": r.id,
            "entity_type": r.entity_type,
            "short_id": getattr(r, "short_id", None),
            "name": r.name,
            "description": r.description or "",
            "metadata_json": r.metadata_json or "{}",
        })
    return out


def load_location_catalog(session, project_id: str) -> list[dict]:
    from app.models.project import EntityCanon
    rows = session.query(EntityCanon).filter(
        EntityCanon.project_id == project_id,
        EntityCanon.entity_type == "location",
    ).order_by(
        EntityCanon.entity_type, EntityCanon.short_id, EntityCanon.id,
    ).all()
    out: list[dict] = []
    for r in rows:
        out.append({
            "canon_id": r.id,
            "entity_type": r.entity_type,
            "short_id": getattr(r, "short_id", None),
            "name": r.name,
            "description": r.description or "",
            "metadata_json": r.metadata_json or "{}",
        })
    return out


# ============================================================================
# SourceBundle builder
# ============================================================================
def _make_source_item(*, source_ref: str, kind: str, text: str,
                      extras: Optional[dict] = None) -> SourceItem:
    """W1c: verbatim source preservation. NO NFC normalize, NO strip, NO
    truncation. sha256 / char_count computed on the exact bytes/chars
    passed in.
    """
    raw = _coerce_source_text(text)
    return SourceItem(
        source_ref=source_ref, kind=kind, text=raw,
        sha256=_sha256_hex(raw), char_count=len(raw),
        extras=dict(extras or {}),
    )


def build_source_bundle(*, spec: SampleFixtureSpec, run_id: str,
                         planning_text: str, episode_text: str,
                         selected_shots: list[dict],
                         entity_catalog: list[dict],
                         location_catalog: list[dict],
                         diagnostic_artifacts: Optional[list[tuple[str, Any]]] = None,
                         missing_inputs: Optional[list[dict]] = None,
                         ) -> SourceBundle:
    sources: list[SourceItem] = []
    sources.append(_make_source_item(
        source_ref=f"planning_doc:{spec.project_id}",
        kind="planning_doc", text=planning_text,
    ))
    sources.append(_make_source_item(
        source_ref=f"episode_fulltext:{spec.episode_id}",
        kind="episode_fulltext", text=episode_text,
    ))
    for sh in selected_shots:
        still_id = sh.get("still_id") or ""
        combined = "\n\n".join(filter(None, [
            sh.get("shot_description") or "",
            sh.get("scene_summary") or "",
            f"visible_entities_json: {sh.get('visible_entities_json', '[]')}",
        ]))
        sources.append(_make_source_item(
            source_ref=f"shot:{still_id}",
            kind="selected_shot", text=combined,
            extras={
                "scene_index": sh.get("scene_index"),
                "shot_index": sh.get("shot_index"),
                "still_id": still_id,
                "loc_short_ids_visible": sh.get("loc_short_ids_visible") or [],
            },
        ))
    for ent in entity_catalog:
        canon_id = ent.get("canon_id") or ""
        body_parts = [
            f"name: {ent.get('name', '')}",
            f"entity_type: {ent.get('entity_type', '')}",
            f"description: {ent.get('description', '')}",
            f"metadata_json: {ent.get('metadata_json', '{}')}",
        ]
        sources.append(_make_source_item(
            source_ref=f"entity:{canon_id}",
            kind="entity_catalog", text="\n".join(body_parts),
            extras={
                "entity_type": ent.get("entity_type"),
                "short_id": ent.get("short_id"),
                "name": ent.get("name"),
            },
        ))
    for loc in location_catalog:
        canon_id = loc.get("canon_id") or ""
        body_parts = [
            f"name: {loc.get('name', '')}",
            f"short_id: {loc.get('short_id', '')}",
            f"description: {loc.get('description', '')}",
            f"metadata_json: {loc.get('metadata_json', '{}')}",
        ]
        sources.append(_make_source_item(
            source_ref=f"location:{canon_id}",
            kind="location_catalog", text="\n".join(body_parts),
            extras={
                "short_id": loc.get("short_id"), "name": loc.get("name"),
            },
        ))
    if diagnostic_artifacts:
        for path, obj in diagnostic_artifacts:
            body = json.dumps(obj, ensure_ascii=False, indent=2, default=str)
            sources.append(_make_source_item(
                source_ref=f"artifact:{path}", kind="existing_artifact",
                text=body, extras={"deprecated_diagnostic": True},
            ))
    return SourceBundle(
        bundle_id=f"bsx_{run_id}",
        fixture_id=spec.fixture_id,
        project_id=spec.project_id,
        episode_id=spec.episode_id,
        location_short_id=spec.location_short_id,
        sources=sources,
        missing_inputs=list(missing_inputs or []),
    )


# ============================================================================
# Schema builders (Stage A + Stage B, W0g)
# ============================================================================
def _evidence_ref_schema() -> dict:
    return {
        "type": "object",
        "required": ["source_ref", "quote", "confidence"],
        "properties": {
            "source_ref": {"type": "string"},
            "quote": {"type": "string"},
            "char_start": {"type": ["integer", "null"]},
            "char_end": {"type": ["integer", "null"]},
            "confidence": {
                "type": "string",
                "enum": list(CONFIDENCE_BANDS),
            },
        },
    }


def build_world_brief_schema(bundle: SourceBundle) -> dict:
    return {
        "schema_version": STAGE_A_SCHEMA_VERSION,
        "bundle_id": bundle.bundle_id,
        "type": "object",
        "required": [
            "schema_version", "bundle_id",
            "world_background_brief", "confidence_band",
        ],
        "forbidden_top_level_or_item_fields": list(TOPOLOGY_FIELDS_FORBIDDEN) + ["count"],
        "confidence_band_enum": list(CONFIDENCE_BANDS),
        "world_background_brief": {
            "type": "object",
            "required": [
                "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",
            ],
            "caps": {
                "era_and_time_period_max_chars": STAGE_A_FIELD_CHAR_CAPS["era_and_time_period"],
                "geographic_cultural_grounding_max_chars": STAGE_A_FIELD_CHAR_CAPS["geographic_cultural_grounding"],
                "technology_and_material_baseline_max_chars": STAGE_A_FIELD_CHAR_CAPS["technology_and_material_baseline"],
                "social_economic_visual_tone_max_chars": STAGE_A_FIELD_CHAR_CAPS["social_economic_visual_tone"],
                "genre_mood_constraints_max_chars": STAGE_A_FIELD_CHAR_CAPS["genre_mood_constraints"],
                "background_relevant_do_not_assume_max_items": STAGE_A_BG_DO_NOT_ASSUME_MAX,
                "background_relevant_do_not_assume_item_max_chars": STAGE_A_BG_DO_NOT_ASSUME_ITEM_CHARS,
                "irrelevant_or_do_not_pass_down_max_items": STAGE_A_IRRELEVANT_MAX,
                "irrelevant_or_do_not_pass_down_item_max_chars": STAGE_A_IRRELEVANT_ITEM_CHARS,
            },
            "evidence_refs": {
                "type": "array", "minItems": 1,
                "items": _evidence_ref_schema(),
            },
        },
    }


def build_minimal_spatial_brief_schema(bundle: SourceBundle) -> dict:
    item_with_evidence = lambda required, props: {  # noqa
        "type": "object",
        "required": list(required) + ["evidence_refs"],
        "properties": {**props, "evidence_refs": {
            "type": "array", "minItems": 1, "items": _evidence_ref_schema(),
        }},
    }
    return {
        "schema_version": STAGE_B_SCHEMA_VERSION,
        "bundle_id": bundle.bundle_id,
        "type": "object",
        "required": [
            "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",
        ],
        "forbidden_top_level_or_item_fields":
            list(TOPOLOGY_FIELDS_FORBIDDEN)
            + ["count", "world_context_brief", "world_background_brief"],
        "world_only_top_level_keys": ["world_brief_ref", "world_hints_for_background"],
        "confidence_band_enum": list(CONFIDENCE_BANDS),
        "base_plate_role_enum": list(BASE_PLATE_ROLES),
        "caps": {
            "world_hints_for_background_max_items": STAGE_B_WORLD_HINTS_MAX,
            "world_hints_for_background_item_max_chars": STAGE_B_WORLD_HINTS_ITEM_CHARS,
            "place_identity_label_max_chars": STAGE_B_PLACE_LABEL_CHARS,
            "place_identity_one_sentence_max_chars": STAGE_B_PLACE_ONE_SENTENCE_CHARS,
            "continuity_groups_max_items": STAGE_B_CONTINUITY_GROUPS_MAX,
            "continuity_group_label_max_chars": STAGE_B_CONTINUITY_GROUP_LABEL_CHARS,
            "continuity_group_list_max_items": STAGE_B_CONTINUITY_GROUP_LIST_MAX,
            "continuity_group_list_item_max_chars": STAGE_B_CONTINUITY_GROUP_ITEM_CHARS,
            "essential_spatial_relations_max_items": STAGE_B_SPATIAL_RELATIONS_MAX,
            "essential_spatial_relation_desc_max_chars": STAGE_B_SPATIAL_RELATION_DESC_CHARS,
            "visual_anchors_max_items": STAGE_B_VISUAL_ANCHORS_MAX,
            "visual_anchor_desc_max_chars": STAGE_B_VISUAL_ANCHOR_DESC_CHARS,
            "state_variations_max_items": STAGE_B_STATE_VARIATIONS_MAX,
            "state_variation_desc_max_chars": STAGE_B_STATE_VARIATION_DESC_CHARS,
            "state_variation_list_max_items": STAGE_B_STATE_VARIATION_LIST_MAX,
            "state_variation_list_item_max_chars": STAGE_B_STATE_VARIATION_ITEM_CHARS,
            "generation_notes_base_plate_brief_max_chars": STAGE_B_GENERATION_NOTES_BASE_PLATE_BRIEF_CHARS,
            "generation_notes_array_max_items": STAGE_B_GENERATION_NOTES_ARRAY_MAX,
            "generation_notes_array_item_max_chars": STAGE_B_GENERATION_NOTES_ARRAY_ITEM_CHARS,
            "slug_max_chars": STAGE_B_SLUG_CHARS,
        },
        "world_brief_ref": {"type": "string"},
        "world_hints_for_background": {
            "type": "array",
            "items": {"type": "string"},
        },
        "place_identity": item_with_evidence(
            ["label", "one_sentence"],
            {
                "label": {"type": "string"},
                "one_sentence": {"type": "string"},
            },
        ),
        "continuity_groups": {
            "type": "array",
            "items": item_with_evidence(
                ["group_id", "label",
                 "what_must_stay_consistent", "allowed_variations"],
                {
                    "group_id": {"type": "string"},
                    "label": {"type": "string"},
                    "what_must_stay_consistent": {
                        "type": "array", "items": {"type": "string"},
                    },
                    "allowed_variations": {
                        "type": "array", "items": {"type": "string"},
                    },
                },
            ),
        },
        "essential_spatial_relations": {
            "type": "array",
            "items": item_with_evidence(
                ["relation_id", "description",
                 "why_it_matters_for_generation"],
                {
                    "relation_id": {"type": "string"},
                    "description": {"type": "string"},
                    "why_it_matters_for_generation": {"type": "string"},
                },
            ),
        },
        "visual_anchors": {
            "type": "array",
            "items": item_with_evidence(
                ["anchor_id", "description", "role", "base_plate_role"],
                {
                    "anchor_id": {"type": "string"},
                    "description": {"type": "string"},
                    "role": {"type": "string"},
                    "base_plate_role": {
                        "type": "string", "enum": list(BASE_PLATE_ROLES),
                    },
                },
            ),
        },
        "state_variations": {
            "type": "array",
            "items": item_with_evidence(
                ["state_id", "description",
                 "changes_only", "must_not_change"],
                {
                    "state_id": {"type": "string"},
                    "description": {"type": "string"},
                    "changes_only": {
                        "type": "array", "items": {"type": "string"},
                    },
                    "must_not_change": {
                        "type": "array", "items": {"type": "string"},
                    },
                },
            ),
        },
        "generation_notes": {
            "type": "object",
            "required": [
                "base_plate_prompt_brief",
                "shot_background_prompt_rules",
                "avoid_over_specification",
                "unknowns_to_keep_loose",
            ],
            "properties": {
                "base_plate_prompt_brief": {"type": "string"},
                "shot_background_prompt_rules": {
                    "type": "array", "items": {"type": "string"},
                },
                "avoid_over_specification": {
                    "type": "array", "items": {"type": "string"},
                },
                "unknowns_to_keep_loose": {
                    "type": "array", "items": {"type": "string"},
                },
            },
        },
        "confidence_band": {
            "type": "string", "enum": list(CONFIDENCE_BANDS),
        },
    }


# ============================================================================
# Evidence Filter schema (W2a)
# ============================================================================
def _brief_evidence_ref_schema() -> dict:
    """W2-simple: evidence_refs carry only provenance (source_ref + quote
    + optional char positions). Confidence sits at the rule level, NOT
    on each evidence_ref."""
    return {
        "type": "object",
        "required": ["source_ref", "quote"],
        "properties": {
            "source_ref": {"type": "string"},
            "quote": {"type": "string"},
            "char_start": {"type": ["integer", "null"]},
            "char_end": {"type": ["integer", "null"]},
        },
    }


def build_background_evidence_pack_schema(bundle: SourceBundle) -> dict:
    """W2-simple BackgroundContinuityBrief schema (Codex consultation
    APPROVED with revisions).

    Note: the function name is kept for source-compatibility; what it
    returns is the BackgroundContinuityBrief schema, NOT the old W2a..
    W2a-f evidence pack.
    """
    er = _brief_evidence_ref_schema()
    rule_with_ev_required = lambda required, props: {  # noqa: E731
        "type": "object",
        "required": list(required) + ["confidence_band", "evidence_refs"],
        "properties": {
            **props,
            "confidence_band": {
                "type": "string", "enum": list(CONFIDENCE_BANDS),
            },
            "evidence_refs": {
                "type": "array", "minItems": 1, "items": er,
            },
        },
    }
    rule_with_ev_optional = lambda required, props: {  # noqa: E731
        # allowed_creative_freedom: evidence_refs is optional, replaced
        # by a `basis` enum that records why this freedom exists.
        "type": "object",
        "required": list(required) + ["basis", "confidence_band"],
        "properties": {
            **props,
            "basis": {
                "type": "string",
                "enum": list(CONTINUITY_BRIEF_ALLOWED_BASIS_ENUM),
            },
            "confidence_band": {
                "type": "string", "enum": list(CONFIDENCE_BANDS),
            },
            "evidence_refs": {
                "type": "array", "items": er,
            },
        },
    }
    return {
        "schema_version": EVIDENCE_PACK_SCHEMA_VERSION,
        "source_bundle_ref": bundle.bundle_id,
        "type": "object",
        "required": [
            "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",
        ],
        "forbidden_top_level_or_item_fields":
            list(TOPOLOGY_FIELDS_FORBIDDEN)
            + list(EVIDENCE_PACK_FORBIDDEN_INVENTORY_KEYS)
            + list(EVIDENCE_PACK_LEGACY_KEYS),
        "confidence_band_enum": list(CONFIDENCE_BANDS),
        "allowed_creative_freedom_basis_enum":
            list(CONTINUITY_BRIEF_ALLOWED_BASIS_ENUM),
        "caps": {
            "list_max_items": CONTINUITY_BRIEF_LIST_MAX,
            "rule_id_max_chars": CONTINUITY_BRIEF_RULE_ID_CHARS,
            "statement_max_chars": CONTINUITY_BRIEF_STATEMENT_CHARS,
            "why_max_chars": CONTINUITY_BRIEF_WHY_CHARS,
            "list_item_max_chars": CONTINUITY_BRIEF_LIST_ITEM_CHARS,
            "shot_conflict_checks_max_items":
                CONTINUITY_BRIEF_SHOT_CHECKS_MAX,
            "identity_summary_max_chars":
                CONTINUITY_BRIEF_IDENTITY_SUMMARY_CHARS,
            "identity_why_max_chars":
                CONTINUITY_BRIEF_IDENTITY_WHY_CHARS,
            "coverage_notes_max_items": EVIDENCE_PACK_COVERAGE_NOTES_MAX,
            "coverage_note_max_chars": EVIDENCE_PACK_COVERAGE_NOTE_CHARS,
            "coverage_consequence_max_chars":
                EVIDENCE_PACK_COVERAGE_CONSEQUENCE_CHARS,
            "rejected_or_irrelevant_summary_max_items":
                EVIDENCE_PACK_REJECTED_MAX,
            "rejected_short_note_max_chars": EVIDENCE_PACK_REJECTED_NOTE_CHARS,
            "rejected_why_not_pass_down_max_chars":
                EVIDENCE_PACK_REJECTED_WHY_CHARS,
        },
        "common_place_identity": {
            "type": "object",
            "required": ["summary", "why_this_identity",
                         "confidence_band", "evidence_refs"],
            "properties": {
                "summary": {"type": "string"},
                "why_this_identity": {"type": "string"},
                "confidence_band": {
                    "type": "string", "enum": list(CONFIDENCE_BANDS),
                },
                "evidence_refs": {
                    "type": "array", "minItems": 1, "items": er,
                },
            },
        },
        "must_stay_consistent": {
            "type": "array",
            "items": rule_with_ev_required(
                ["rule_id", "statement", "why_consistent"],
                {
                    "rule_id": {"type": "string"},
                    "statement": {"type": "string"},
                    "why_consistent": {"type": "string"},
                },
            ),
        },
        "must_not_contradict": {
            "type": "array",
            "items": rule_with_ev_required(
                ["rule_id", "do_not_introduce", "why"],
                {
                    "rule_id": {"type": "string"},
                    "do_not_introduce": {"type": "string"},
                    "why": {"type": "string"},
                },
            ),
        },
        "allowed_creative_freedom": {
            "type": "array",
            "items": rule_with_ev_optional(
                ["rule_id", "free_to_choose", "guidance"],
                {
                    "rule_id": {"type": "string"},
                    "free_to_choose": {"type": "string"},
                    "guidance": {"type": "string"},
                },
            ),
        },
        "state_change_rules": {
            "type": "array",
            "items": rule_with_ev_required(
                ["rule_id", "change_kind", "when_applies",
                 "what_changes", "what_stays"],
                {
                    "rule_id": {"type": "string"},
                    "change_kind": {"type": "string"},
                    "when_applies": {"type": "string"},
                    "what_changes": {
                        "type": "array", "items": {"type": "string"},
                    },
                    "what_stays": {
                        "type": "array", "items": {"type": "string"},
                    },
                },
            ),
        },
        "shot_conflict_checks": {
            "type": "array",
            "items": rule_with_ev_required(
                ["shot_ref", "must_support", "avoid_contradiction"],
                {
                    "shot_ref": {"type": "string"},
                    "must_support": {
                        "type": "array", "items": {"type": "string"},
                    },
                    "avoid_contradiction": {
                        "type": "array", "items": {"type": "string"},
                    },
                },
            ),
        },
        "unknowns_left_to_art_direction": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["area", "note"],
                "properties": {
                    "area": {"type": "string"},
                    "note": {"type": "string"},
                },
            },
        },
        "coverage_notes": {
            "type": "array",
            "items": {
                "type": "object",
                "required": [
                    "note", "consequence_for_background", "confidence_band",
                ],
                "properties": {
                    "note": {"type": "string"},
                    "consequence_for_background": {"type": "string"},
                    "confidence_band": {
                        "type": "string", "enum": list(CONFIDENCE_BANDS),
                    },
                },
            },
        },
        "rejected_or_irrelevant_summary": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["short_note", "why_not_pass_down"],
                "properties": {
                    "short_note": {"type": "string"},
                    "why_not_pass_down": {"type": "string"},
                },
            },
        },
    }


# ============================================================================
# Prompt builders — 5 functions (evidence_filter + 2-stage). Prompt templates
# instruct the LLM directly; §6-A guards apply to *function bodies*
# (signatures + AST) only, not prompt template strings (plan §6-A 검사 범위).
#
# Evidence Filter builder is the ONLY prompt that receives the raw
# SourceBundle. Downstream (world + spatial) builders receive only the
# filtered BackgroundEvidencePack — never `bundle` or raw `source.text`.
#
# Stage B prompt builder signature MUST NOT receive `world_background_brief`
# or `irrelevant_or_do_not_pass_down` parameters — that is enforced by test
# 6-E-22 (W0g BLOCKING).
# ============================================================================
def build_prompt_evidence_filter_system(spec: SampleFixtureSpec) -> str:
    return (
        "You are a background continuity-brief author for a film / "
        "drama production.\n\n"
        "Your goal is NOT an exact spatial decomposition or an "
        "inventory of the location. Your goal is a short constraint "
        "brief that downstream shot generation must not contradict, "
        "and that leaves room for the art-direction team to fill in "
        "the rest. The downstream world / spatial stages receive ONLY "
        "this brief — never the raw SourceBundle.\n\n"
        "★ Every rule statement is a CONSTRAINT STATEMENT, not an "
        "image command. Do NOT use render / depict / show / place / "
        "add — instead say 'keep…', 'do not introduce…', 'free to "
        "choose…', 'when X happens, Y may change while Z stays the "
        "same'.\n"
        "★ Do NOT invent place-specific rules. If the source does not "
        "say it, leave it to art direction (`unknowns_left_to_art_"
        "direction` or `allowed_creative_freedom`).\n"
        "★ Do NOT enumerate doors / windows / furniture counts.\n"
        "★ common_place_identity and must_stay_consistent describe "
        "the PLACE/BACKGROUND only. Do not embed character names, "
        "ownership ('X's home'), plot chronology, or incident labels "
        "('before / after the crime') unless the source-quoted detail "
        "directly changes a visible background constraint. Plot, "
        "character relationships, and event labels belong in "
        "rejected_or_irrelevant_summary.\n\n"
        "Output a single BackgroundContinuityBrief JSON object with "
        "the following top-level keys:\n"
        "  - common_place_identity : a single shared baseline that "
        "every shot's background must respect. Fields: summary "
        "(≤240 chars), why_this_identity (≤180 chars), "
        "confidence_band, evidence_refs (≥1).\n"
        "  - must_stay_consistent : the few constraints that must "
        "hold across every shot of this background. Each item: "
        "rule_id (slug), statement (≤200 chars), why_consistent "
        "(≤180 chars), confidence_band, evidence_refs (≥1).\n"
        "  - must_not_contradict  : what downstream generation must "
        "NOT introduce, because it would clash with the source. Each "
        "item: rule_id, do_not_introduce, why, confidence_band, "
        "evidence_refs (≥1).\n"
        "  - allowed_creative_freedom : explicit zones the art-"
        "direction team may fill freely without breaking the brief. "
        "Each item: rule_id, free_to_choose, guidance, basis (one of "
        "`not_specified` / `weakly_constrained` / "
        "`art_direction_choice`), confidence_band, evidence_refs "
        "(OPTIONAL — these freedoms exist precisely because the "
        "source under-specifies; do not invent evidence to justify "
        "them).\n"
        "  - state_change_rules : how the SAME background may "
        "legitimately vary across scenes/events. Each item: "
        "rule_id, change_kind (free string — e.g. damage, temporary "
        "markings, cleaned state, lighting shift, weather, "
        "crowding), when_applies (free string), what_changes (list), "
        "what_stays (list), confidence_band, evidence_refs (≥1).\n"
        "  - shot_conflict_checks : per-shot constraints. Each "
        "item: shot_ref (MUST be exactly the source_ref of a "
        "selected_shot present in the SourceBundle — never a "
        "free-form label), must_support (list), avoid_contradiction "
        "(list), confidence_band, evidence_refs (≥1). At least ONE "
        "evidence_ref must cite the SAME shot_ref as the rule (i.e. "
        "`source_ref == shot_ref`) — episode-level or planning-doc "
        "quotes alone are not enough to ground a per-shot conflict "
        "check. Additional episode/planning quotes may be included, "
        "but the per-shot evidence is mandatory. You do not have to "
        "write one entry per shot; only write checks where the shot "
        "adds a constraint or risk.\n"
        "  - unknowns_left_to_art_direction : background details the "
        "source does not specify; the art-direction team designs "
        "these. Each item: area, note. No evidence_refs.\n"
        "  - coverage_notes : known gaps in the source that affect "
        "background generation. Each item: note, "
        "consequence_for_background, confidence_band. No evidence_refs.\n"
        "  - rejected_or_irrelevant_summary : plot / character / lore "
        "material you intentionally leave out. Each item: short_note, "
        "why_not_pass_down. No evidence_refs.\n\n"
        "evidence_ref shape: source_ref (must match a SourceBundle "
        "source) + quote (verbatim contiguous substring of that "
        "source's text). Do NOT add `confidence_band` to evidence_refs "
        "— confidence belongs on the rule, because it is the "
        "confidence in the constraint, not in the quote.\n\n"
        "confidence_band (at rule level + common_place_identity + "
        "coverage_notes) must be exactly one of: trusted, plausible, "
        "weak, unknown. Do not use high / medium / low — those tokens "
        "are rejected by the deterministic checker.\n\n"
        "The deterministic checker validates schema, caps, "
        "evidence-ref provenance, slugs, and enums. It does NOT judge "
        "whether a rule is the 'right' rule semantically. Keep the "
        "brief small: each list ≤ 8 items; shot_conflict_checks ≤ "
        "24. Favor a few firm rules over many soft ones.\n\n"
        "Reason semantically. Do not use pattern / term / boundary / "
        "regex / substring / particle thinking.\n"
    )


def build_prompt_evidence_filter_user(*, spec: SampleFixtureSpec,
                                       bundle: SourceBundle,
                                       schema: dict) -> str:
    lines: list[str] = []
    lines.append(f"Bundle id: {bundle.bundle_id}")
    lines.append(f"Fixture: {bundle.fixture_id}")
    lines.append(f"Location short id: {bundle.location_short_id}")
    lines.append(f"Total sources: {len(bundle.sources)}")
    lines.append(f"Char budget estimate: {bundle.char_budget_estimate}")
    lines.append("")
    lines.append("BackgroundContinuityBrief top-level keys:")
    for k in schema.get("required", []):
        lines.append(f"  - {k}")
    lines.append("")
    lines.append(
        "BackgroundContinuityBrief top-level keys: "
        + ", ".join(schema.get("required", []))
    )
    lines.append(
        "allowed_creative_freedom basis enum: "
        + ", ".join(
            schema.get("allowed_creative_freedom_basis_enum", []))
    )
    lines.append(
        "confidence_band enum (exactly one of, no synonyms): "
        + ", ".join(schema.get("confidence_band_enum", []))
        + " — do not emit high / medium / low."
    )
    lines.append("")
    lines.append(
        "Caps (deterministic): each rule list ≤ "
        f"{CONTINUITY_BRIEF_LIST_MAX}; shot_conflict_checks ≤ "
        f"{CONTINUITY_BRIEF_SHOT_CHECKS_MAX}; statement ≤ "
        f"{CONTINUITY_BRIEF_STATEMENT_CHARS} chars; why ≤ "
        f"{CONTINUITY_BRIEF_WHY_CHARS} chars; list item ≤ "
        f"{CONTINUITY_BRIEF_LIST_ITEM_CHARS} chars; common_place_"
        f"identity.summary ≤ {CONTINUITY_BRIEF_IDENTITY_SUMMARY_CHARS}."
    )
    lines.append(
        "★ Every rule statement is a CONSTRAINT statement, not an "
        "image command. ★ shot_ref must EXACTLY equal a "
        "selected_shot source_ref from the SourceBundle. "
        "★ Do not add confidence_band to evidence_refs — it lives on "
        "the rule. ★ allowed_creative_freedom items may omit "
        "evidence_refs; record a `basis` enum instead."
    )
    lines.append("")
    lines.append("===== BEGIN SOURCE BUNDLE =====")
    for src in bundle.sources:
        lines.append("")
        lines.append(f"[[source_ref]] {src.source_ref}")
        lines.append(f"[[kind]] {src.kind}")
        lines.append(f"[[sha256]] {src.sha256}")
        lines.append(f"[[char_count]] {src.char_count}")
        if src.extras:
            lines.append(
                f"[[extras]] {json.dumps(src.extras, ensure_ascii=False)}"
            )
        lines.append("[[text]]")
        lines.append(src.text)
    lines.append("")
    lines.append("===== END SOURCE BUNDLE =====")
    lines.append("")
    lines.append("Output JSON only.")
    return "\n".join(lines)


# ----------------------------------------------------------------------------
# Evidence Filter correction prompt (W2a-d, Codex 권장).
#
# Triggered ONLY when the initial Evidence Filter pack failed
# deterministic validation with `quote_not_in_source`. Same schema, same
# bundle; LLM is asked to fix the failing items so every quote is a
# verbatim substring of its cited source.text. Deterministic code does
# NOT modify quotes itself; this is purely an instruction to the LLM.
# ----------------------------------------------------------------------------
def build_prompt_evidence_filter_correction_system(
        spec: SampleFixtureSpec) -> str:
    return (
        "You are a correction pass for a BackgroundContinuityBrief "
        "that failed deterministic validation.\n\n"
        "The previous pack you produced had one or more evidence_items "
        "whose `quote` field is NOT a verbatim substring of the cited "
        "source's `text`. The deterministic checker rejected those items "
        "with `evidence_filter:quote_not_in_source:<index>`.\n\n"
        "Re-emit the SAME BackgroundContinuityBrief JSON object, keeping "
        "valid items unchanged. For each failing item, copy a quote "
        "EXACTLY from the cited source.text — a verbatim contiguous "
        "substring, character for character, including spacing and "
        "punctuation. Do not paraphrase, do not translate, do not "
        "summarize, do not add or remove any character. If no usable "
        "verbatim quote exists in the cited source, drop the item "
        "entirely (do not invent a quote).\n\n"
        "Schema, caps, enums, and forbidden-key rules are identical "
        "to the initial pass — this is the W2-simple "
        "BackgroundContinuityBrief: common_place_identity, "
        "must_stay_consistent, must_not_contradict, "
        "allowed_creative_freedom (with a `basis` enum), "
        "state_change_rules, shot_conflict_checks (shot_ref MUST "
        "exact-equal a selected_shot source_ref in the SourceBundle), "
        "unknowns_left_to_art_direction, coverage_notes, "
        "rejected_or_irrelevant_summary. Do NOT reintroduce legacy "
        "keys (`use_for`, `pass_down_hint`, `why_background_relevant`, "
        "`role`, `applies_to`, `background_fact`, `why_keep`, "
        "`evidence_items`). confidence_band must be exactly one of: "
        "trusted, plausible, weak, unknown. Do not use high / medium "
        "/ low. Do NOT add confidence_band to evidence_refs — "
        "confidence sits on the rule. Do not inventory objects, do "
        "not introduce topology keys, do not add `count`. Every "
        "rule's statement is a CONSTRAINT, not an image-rendering "
        "command. Output a single JSON object only.\n"
    )


def build_prompt_evidence_filter_correction_user(
        *, spec: SampleFixtureSpec, bundle: SourceBundle, schema: dict,
        invalid_pack: dict, failed_checks: list[str]) -> str:
    """Correction user prompt — receives the FULL SourceBundle (no
    truncation), the invalid pack, and the list of failed_checks."""
    lines: list[str] = []
    lines.append(f"Bundle id: {bundle.bundle_id}")
    lines.append(f"Fixture: {bundle.fixture_id}")
    lines.append(f"Location short id: {bundle.location_short_id}")
    lines.append(f"Total sources: {len(bundle.sources)}")
    lines.append(f"Char budget estimate: {bundle.char_budget_estimate}")
    lines.append("")
    lines.append("Failed checks from the deterministic validator:")
    for fc in failed_checks:
        lines.append(f"  - {fc}")
    lines.append("")
    lines.append(
        "Rejected evidence_refs by (section, index) extracted from "
        "failed_checks (quote_not_in_source entries — format "
        "`evidence_filter:quote_not_in_source:<section>:<rule_index>`):"
    )
    failed_locations: list[tuple[str, int]] = []
    for fc in failed_checks:
        if "quote_not_in_source:" in fc:
            tail = fc.split("quote_not_in_source:")[-1]
            parts = tail.split(":")
            if len(parts) >= 2:
                try:
                    failed_locations.append((parts[0], int(parts[1])))
                except ValueError:
                    pass
    for section, idx in sorted(set(failed_locations)):
        items_in_section = invalid_pack.get(section) or []
        if section == "common_place_identity":
            cpi = invalid_pack.get("common_place_identity") or {}
            refs = cpi.get("evidence_refs") or []
            if 0 <= idx < len(refs):
                lines.append(
                    f"  [{section}:{idx}] source_ref="
                    f"{refs[idx].get('source_ref', '')!r} "
                    f"rejected_quote={refs[idx].get('quote', '')!r}"
                )
            continue
        if 0 <= idx < len(items_in_section) and isinstance(
                items_in_section[idx], dict):
            rule = items_in_section[idx]
            lines.append(
                f"  [{section}:{idx}] rule_id="
                f"{rule.get('rule_id', rule.get('shot_ref', ''))!r}"
            )
    lines.append("")
    lines.append("Your previous brief (verbatim):")
    lines.append(json.dumps(invalid_pack, ensure_ascii=False, indent=2))
    lines.append("")
    lines.append(
        "BackgroundContinuityBrief top-level required keys: "
        + ", ".join(schema.get("required", []))
    )
    lines.append(
        "allowed_creative_freedom basis enum: "
        + ", ".join(
            schema.get("allowed_creative_freedom_basis_enum", []))
    )
    lines.append(
        "confidence_band enum: "
        + ", ".join(schema.get("confidence_band_enum", []))
        + " — do not emit high / medium / low."
    )
    lines.append(
        "evidence_ref shape: source_ref + verbatim quote (no "
        "confidence_band on evidence_ref). shot_ref must EXACTLY "
        "equal a selected_shot source_ref from the SourceBundle."
    )
    lines.append("")
    lines.append("===== BEGIN SOURCE BUNDLE (verbatim, do not edit) =====")
    for src in bundle.sources:
        lines.append("")
        lines.append(f"[[source_ref]] {src.source_ref}")
        lines.append(f"[[kind]] {src.kind}")
        lines.append(f"[[sha256]] {src.sha256}")
        lines.append(f"[[char_count]] {src.char_count}")
        if src.extras:
            lines.append(
                f"[[extras]] {json.dumps(src.extras, ensure_ascii=False)}"
            )
        lines.append("[[text]]")
        lines.append(src.text)
    lines.append("")
    lines.append("===== END SOURCE BUNDLE =====")
    lines.append("")
    lines.append(
        "Output the corrected BackgroundContinuityBrief JSON object only."
    )
    return "\n".join(lines)


# ============================================================================
# Downstream prompt builders (W2a: receive ONLY the evidence_pack).
# ============================================================================
def build_prompt_world_system(spec: SampleFixtureSpec) -> str:
    return (
        "You are a background world grounding assistant.\n\n"
        "You will receive a filtered BackgroundEvidencePack from the "
        "upstream Evidence Filter stage. The raw SourceBundle is NOT "
        "available at this stage — only the pack's evidence_items, "
        "coverage_notes, and pass_down_hints. Produce a single "
        "BackgroundWorldBrief JSON object that captures only the world "
        "grounding needed to keep T2I background generation from defaulting "
        "to luxury, Western, futuristic, or fantasy aesthetics.\n\n"
        "Extract only background-generation-relevant world grounding. "
        "Do not pass plot lore unless it changes visual background "
        "assumptions. Put plot facts, character relationships, mythology "
        "lore, and event chronology that do NOT change visual background "
        "into `irrelevant_or_do_not_pass_down` so the downstream stage "
        "skips them.\n\n"
        "Each world field is short (cap enforced by the deterministic "
        "checker). Use 'unknown' if the pack does not say. Every claim "
        "must carry an evidence_ref whose source_ref + quote are taken "
        "from a matching evidence item in the filtered "
        "BackgroundEvidencePack.\n\n"
        "Reason semantically. Do not use pattern / term / boundary / regex / "
        "substring / particle thinking. Speak in plain semantic terms a "
        "production designer would understand.\n"
    )


def build_prompt_world_user(*, spec: SampleFixtureSpec,
                             evidence_pack: dict, schema: dict) -> str:
    """Stage A user prompt builder (W2a).

    SIGNATURE GUARD: receives ONLY the filtered BackgroundEvidencePack —
    never the raw SourceBundle, never raw `source.text` or `selected_shots`.
    Tests 6-G `test_world_prompt_builder_does_not_accept_bundle_or_raw_
    sources` enforces this via inspect.signature.
    """
    lines: list[str] = []
    lines.append(
        f"Source bundle ref: {evidence_pack.get('source_bundle_ref', '')}"
    )
    items = evidence_pack.get("evidence_items") or []
    lines.append(f"Filtered evidence items: {len(items)}")
    lines.append("")
    lines.append("Stage A schema top-level keys:")
    for k in schema.get("required", []):
        lines.append(f"  - {k}")
    lines.append("")
    lines.append("world_background_brief required subfields:")
    for k in schema.get("world_background_brief", {}).get("required", []):
        lines.append(f"  - {k}")
    lines.append("")
    lines.append(
        "Per-evidence_ref required fields: source_ref (must equal an evidence "
        "item's source_ref), quote (verbatim from the cited source.text via "
        "the pack), confidence. char_start/char_end optional."
    )
    lines.append("")
    lines.append(
        "===== BEGIN BACKGROUND CONTINUITY BRIEF "
        "(verbatim JSON) ====="
    )
    lines.append(
        json.dumps(evidence_pack, ensure_ascii=False, indent=2)
    )
    lines.append("===== END BACKGROUND CONTINUITY BRIEF =====")
    lines.append("")
    lines.append(
        "Use the brief as constraint context. `rejected_or_irrelevant_"
        "summary` is a CAUTION reference only — do not turn its "
        "entries into image instructions."
    )
    lines.append("Output JSON only.")
    return "\n".join(lines)


def build_prompt_spatial_system(spec: SampleFixtureSpec) -> str:
    return (
        "You are a minimal spatial brief assistant for T2I/I2I background "
        "generation.\n\n"
        "You will receive a filtered BackgroundEvidencePack from the "
        "upstream Evidence Filter stage plus the short world hints from "
        "Stage A. The raw SourceBundle is NOT available at this stage. "
        "Produce a single MinimalSpatialBrief JSON object.\n\n"
        "Use only the compact world hints from Stage A, not the full source "
        "world analysis. Do not re-extract plot or lore that Stage A marked "
        "as irrelevant.\n\n"
        "Keep the brief compact. Avoid over specification. Do not inventory "
        "objects or count doors / windows / furniture. Do not assume the "
        "location is a house, room, set, building, or interior. Extract "
        "spatial decomposition that works for any place type — coastal "
        "village, vessel, forest, alley, ritual site, etc.\n\n"
        "Separate what must stay consistent across shots from what may vary "
        "by state or camera distance. Treat the base plate as an identity / "
        "scale / layout-feel anchor — not as the literal rendering of every "
        "shot.\n\n"
        "Caps are enforced by the deterministic checker: continuity_groups "
        "≤ 3; essential_spatial_relations ≤ 5; visual_anchors ≤ 5; "
        "state_variations ≤ 5; world_hints_for_background ≤ 5. Each "
        "description string is short.\n\n"
        "Every evidence_ref must use verbatim quote and a source_ref taken "
        "from a matching evidence item in the filtered "
        "BackgroundEvidencePack. Reason semantically; do not use pattern / "
        "regex / boundary / particle thinking.\n"
    )


def build_prompt_spatial_user(*, spec: SampleFixtureSpec,
                               evidence_pack: dict, schema: dict,
                               world_brief_ref: str,
                               world_hints_for_background: list[str]) -> str:
    """Stage B user prompt builder (plan §6-E-22 W0g BLOCKING + W2a).

    SIGNATURE GUARD: receives ONLY the filtered BackgroundEvidencePack and
    the short Stage A hints. MUST NOT receive `bundle`, raw
    `source.text`, `world_background_brief`, or
    `irrelevant_or_do_not_pass_down`. Tests
    `test_prompts_separate_*_spatial_builder_excludes_full_stage_a` and
    6-G `test_spatial_prompt_builder_does_not_accept_bundle_or_raw_sources`
    verify via inspect.signature.
    """
    lines: list[str] = []
    lines.append(
        f"Source bundle ref: {evidence_pack.get('source_bundle_ref', '')}"
    )
    items = evidence_pack.get("evidence_items") or []
    lines.append(f"Filtered evidence items: {len(items)}")
    lines.append("")
    lines.append(f"world_brief_ref: {world_brief_ref}")
    lines.append("world_hints_for_background:")
    for h in world_hints_for_background:
        lines.append(f"  - {h}")
    lines.append("")
    lines.append("Stage B schema top-level keys:")
    for k in schema.get("required", []):
        lines.append(f"  - {k}")
    lines.append("")
    lines.append(
        "Per-evidence_ref required fields: source_ref (must equal an "
        "evidence item's source_ref), quote (verbatim from the cited "
        "source.text via the pack), confidence. char_start/char_end "
        "optional."
    )
    lines.append("")
    lines.append(
        "===== BEGIN BACKGROUND CONTINUITY BRIEF "
        "(verbatim JSON) ====="
    )
    lines.append(
        json.dumps(evidence_pack, ensure_ascii=False, indent=2)
    )
    lines.append("===== END BACKGROUND CONTINUITY BRIEF =====")
    lines.append("")
    lines.append(
        "Use the brief as constraint context. `rejected_or_irrelevant_"
        "summary` is a CAUTION reference only — do not turn its "
        "entries into image instructions."
    )
    lines.append("Output JSON only.")
    return "\n".join(lines)


# ============================================================================
# Deterministic checker — 10 항목 (plan §3). Provenance + shape + cap +
# structural input contract. NO content scan, NO substring/regex meaning.
#
# This function's body MUST NOT scan reasoning_basis or brief description
# text via `in` / `.find(` / `.contains(` / `.lower()` / `for word in
# deny_list`. test 6-A-4 enforces this via AST.
# ============================================================================
def _iter_evidence_refs_stage_a(brief: dict):
    wbb = brief.get("world_background_brief") or {}
    for er in (wbb.get("evidence_refs") or []):
        yield ("world_background_brief", er)


def _iter_evidence_refs_stage_b(brief: dict):
    place = brief.get("place_identity") or {}
    for er in (place.get("evidence_refs") or []):
        yield ("place_identity", er)
    for kind in ("continuity_groups", "essential_spatial_relations",
                  "visual_anchors", "state_variations"):
        for item in (brief.get(kind) or []):
            if not isinstance(item, dict):
                continue
            for er in (item.get("evidence_refs") or []):
                yield (kind, er)


def _check_evidence_refs(*, brief: dict, bundle: SourceBundle,
                         iter_fn, fails: list[str],
                         augmented: list[dict], stage_label: str) -> None:
    source_text_by_ref = {s.source_ref: s.text for s in bundle.sources}
    for owner_label, er in iter_fn(brief):
        if not isinstance(er, dict):
            fails.append(f"{stage_label}:{owner_label}:evidence_ref_not_dict")
            continue
        sr = er.get("source_ref")
        if sr not in source_text_by_ref:
            fails.append(
                f"{stage_label}:{owner_label}:source_ref_unresolved:{sr!r}"
            )
            continue
        conf = er.get("confidence")
        if conf not in CONFIDENCE_BANDS:
            fails.append(
                f"{stage_label}:{owner_label}:confidence_invalid:{conf!r}"
            )
        q = er.get("quote") or ""
        if not q or not isinstance(q, str):
            fails.append(f"{stage_label}:{owner_label}:quote_missing")
            continue
        idx = source_text_by_ref[sr].find(q)
        if idx < 0:
            fails.append(f"{stage_label}:{owner_label}:quote_not_in_source")
            continue
        augmented.append({
            "stage": stage_label,
            "owner": owner_label,
            "source_ref": sr,
            "char_start": idx,
            "char_end": idx + len(q),
        })


def _check_required(brief: dict, required_keys: list[str],
                     stage_label: str, fails: list[str]) -> None:
    for k in required_keys:
        if k not in brief:
            fails.append(f"{stage_label}:missing_top_level_key:{k}")


def _check_caps_stage_a(brief: dict, fails: list[str]) -> None:
    wbb = brief.get("world_background_brief") or {}
    for field_name, cap in STAGE_A_FIELD_CHAR_CAPS.items():
        val = wbb.get(field_name)
        if isinstance(val, str) and len(val) > cap:
            fails.append(
                f"stage_a:cap_exceeded:{field_name}:{len(val)}>{cap}"
            )
    bg = wbb.get("background_relevant_do_not_assume") or []
    if isinstance(bg, list):
        if len(bg) > STAGE_A_BG_DO_NOT_ASSUME_MAX:
            fails.append(
                f"stage_a:cap_exceeded:bg_do_not_assume_items:"
                f"{len(bg)}>{STAGE_A_BG_DO_NOT_ASSUME_MAX}"
            )
        for item in bg:
            if isinstance(item, str) and len(item) > STAGE_A_BG_DO_NOT_ASSUME_ITEM_CHARS:
                fails.append(
                    f"stage_a:cap_exceeded:bg_do_not_assume_item:"
                    f"{len(item)}>{STAGE_A_BG_DO_NOT_ASSUME_ITEM_CHARS}"
                )
    irr = wbb.get("irrelevant_or_do_not_pass_down") or []
    if isinstance(irr, list):
        if len(irr) > STAGE_A_IRRELEVANT_MAX:
            fails.append(
                f"stage_a:cap_exceeded:irrelevant_items:"
                f"{len(irr)}>{STAGE_A_IRRELEVANT_MAX}"
            )
        for item in irr:
            if isinstance(item, str) and len(item) > STAGE_A_IRRELEVANT_ITEM_CHARS:
                fails.append(
                    f"stage_a:cap_exceeded:irrelevant_item:"
                    f"{len(item)}>{STAGE_A_IRRELEVANT_ITEM_CHARS}"
                )


def _check_caps_stage_b(brief: dict, fails: list[str]) -> None:
    pid = brief.get("place_identity") or {}
    lbl = pid.get("label") or ""
    if isinstance(lbl, str) and len(lbl) > STAGE_B_PLACE_LABEL_CHARS:
        fails.append(
            f"stage_b:cap_exceeded:place_label:{len(lbl)}>{STAGE_B_PLACE_LABEL_CHARS}"
        )
    one = pid.get("one_sentence") or ""
    if isinstance(one, str) and len(one) > STAGE_B_PLACE_ONE_SENTENCE_CHARS:
        fails.append(
            f"stage_b:cap_exceeded:place_one_sentence:{len(one)}>{STAGE_B_PLACE_ONE_SENTENCE_CHARS}"
        )
    wh = brief.get("world_hints_for_background") or []
    if isinstance(wh, list):
        if len(wh) > STAGE_B_WORLD_HINTS_MAX:
            fails.append(
                f"stage_b:cap_exceeded:world_hints_items:"
                f"{len(wh)}>{STAGE_B_WORLD_HINTS_MAX}"
            )
        for item in wh:
            if isinstance(item, str) and len(item) > STAGE_B_WORLD_HINTS_ITEM_CHARS:
                fails.append(
                    f"stage_b:cap_exceeded:world_hints_item:"
                    f"{len(item)}>{STAGE_B_WORLD_HINTS_ITEM_CHARS}"
                )
    for kind, max_items, desc_cap in (
        ("continuity_groups", STAGE_B_CONTINUITY_GROUPS_MAX, None),
        ("essential_spatial_relations", STAGE_B_SPATIAL_RELATIONS_MAX,
         STAGE_B_SPATIAL_RELATION_DESC_CHARS),
        ("visual_anchors", STAGE_B_VISUAL_ANCHORS_MAX,
         STAGE_B_VISUAL_ANCHOR_DESC_CHARS),
        ("state_variations", STAGE_B_STATE_VARIATIONS_MAX,
         STAGE_B_STATE_VARIATION_DESC_CHARS),
    ):
        items = brief.get(kind) or []
        if isinstance(items, list) and len(items) > max_items:
            fails.append(
                f"stage_b:cap_exceeded:{kind}_items:{len(items)}>{max_items}"
            )
        if desc_cap and isinstance(items, list):
            for item in items:
                d = item.get("description") if isinstance(item, dict) else None
                if isinstance(d, str) and len(d) > desc_cap:
                    fails.append(
                        f"stage_b:cap_exceeded:{kind}_description:"
                        f"{len(d)}>{desc_cap}"
                    )
    notes = brief.get("generation_notes") or {}
    bpb = notes.get("base_plate_prompt_brief") or ""
    if isinstance(bpb, str) and len(bpb) > STAGE_B_GENERATION_NOTES_BASE_PLATE_BRIEF_CHARS:
        fails.append(
            f"stage_b:cap_exceeded:base_plate_prompt_brief:"
            f"{len(bpb)}>{STAGE_B_GENERATION_NOTES_BASE_PLATE_BRIEF_CHARS}"
        )
    for arr_key in ("shot_background_prompt_rules",
                     "avoid_over_specification",
                     "unknowns_to_keep_loose"):
        arr = notes.get(arr_key) or []
        if isinstance(arr, list):
            if len(arr) > STAGE_B_GENERATION_NOTES_ARRAY_MAX:
                fails.append(
                    f"stage_b:cap_exceeded:{arr_key}_items:"
                    f"{len(arr)}>{STAGE_B_GENERATION_NOTES_ARRAY_MAX}"
                )
            for item in arr:
                if isinstance(item, str) and len(item) > STAGE_B_GENERATION_NOTES_ARRAY_ITEM_CHARS:
                    fails.append(
                        f"stage_b:cap_exceeded:{arr_key}_item:"
                        f"{len(item)}>{STAGE_B_GENERATION_NOTES_ARRAY_ITEM_CHARS}"
                    )


def _check_no_forbidden_keys(*, obj: Any, stage_label: str,
                              forbidden: list[str], path: str,
                              fails: list[str]) -> None:
    """Walk dicts only — forbidden KEY name detection. No content scan."""
    if isinstance(obj, dict):
        for k, v in obj.items():
            if k in forbidden:
                fails.append(
                    f"{stage_label}:forbidden_key:{path}:{k}"
                )
            _check_no_forbidden_keys(
                obj=v, stage_label=stage_label, forbidden=forbidden,
                path=f"{path}.{k}", fails=fails,
            )
    elif isinstance(obj, list):
        for i, item in enumerate(obj):
            _check_no_forbidden_keys(
                obj=item, stage_label=stage_label, forbidden=forbidden,
                path=f"{path}[{i}]", fails=fails,
            )


def _check_stage_b_input_contract(*, stage_b_input: dict,
                                   fails: list[str]) -> None:
    """plan §3 #10 (W0g): structural input contract — Stage B input dict's
    world-related keys must be exactly {world_brief_ref, world_hints_for_
    background}. NO content/substring comparison."""
    world_keys = [k for k in stage_b_input.keys()
                  if isinstance(k, str) and k.lower().startswith("world")]
    allowed = {"world_brief_ref", "world_hints_for_background"}
    if set(world_keys) != allowed:
        fails.append(
            f"stage_b_input_contract:world_keys_must_be_exactly_"
            f"{sorted(allowed)}_got:{sorted(world_keys)}"
        )
    if "world_background_brief" in stage_b_input:
        fails.append("stage_b_input_contract:world_background_brief_embedded")
    if "irrelevant_or_do_not_pass_down" in stage_b_input:
        fails.append(
            "stage_b_input_contract:irrelevant_list_in_stage_b_input"
        )


def run_validation(*, world_brief: dict, minimal_brief: dict,
                    stage_b_input: dict, bundle: SourceBundle,
                    spec: SampleFixtureSpec,
                    world_schema: dict,
                    minimal_schema: dict) -> dict:
    """Run all 10 checks. Returns dict with passed/failed_checks/
    augmented_occurrences/summary."""
    fails: list[str] = []
    augmented: list[dict] = []
    _check_required(
        world_brief, world_schema.get("required", []),
        "stage_a", fails,
    )
    if "world_background_brief" in world_brief:
        wbb = world_brief["world_background_brief"]
        if isinstance(wbb, dict):
            for k in world_schema["world_background_brief"]["required"]:
                if k not in wbb:
                    fails.append(f"stage_a:wbb_missing:{k}")
    _check_required(
        minimal_brief, minimal_schema.get("required", []),
        "stage_b", fails,
    )
    _check_evidence_refs(
        brief=world_brief, bundle=bundle,
        iter_fn=_iter_evidence_refs_stage_a, fails=fails,
        augmented=augmented, stage_label="stage_a",
    )
    _check_evidence_refs(
        brief=minimal_brief, bundle=bundle,
        iter_fn=_iter_evidence_refs_stage_b, fails=fails,
        augmented=augmented, stage_label="stage_b",
    )
    a_cb = world_brief.get("confidence_band")
    if a_cb not in CONFIDENCE_BANDS:
        fails.append(f"stage_a:confidence_band_invalid:{a_cb!r}")
    b_cb = minimal_brief.get("confidence_band")
    if b_cb not in CONFIDENCE_BANDS:
        fails.append(f"stage_b:confidence_band_invalid:{b_cb!r}")
    _check_no_forbidden_keys(
        obj=world_brief, stage_label="stage_a",
        forbidden=world_schema.get("forbidden_top_level_or_item_fields", []),
        path="$", fails=fails,
    )
    _check_no_forbidden_keys(
        obj=minimal_brief, stage_label="stage_b",
        forbidden=minimal_schema.get("forbidden_top_level_or_item_fields", []),
        path="$", fails=fails,
    )
    _check_caps_stage_a(world_brief, fails)
    _check_caps_stage_b(minimal_brief, fails)
    _check_stage_b_input_contract(stage_b_input=stage_b_input, fails=fails)
    return {
        "executed": True,
        "passed": len(fails) == 0,
        "failed_checks": fails,
        "augmented_occurrences": augmented,
        "summary": {
            "stage_a_evidence_refs_seen":
                sum(1 for _ in _iter_evidence_refs_stage_a(world_brief)),
            "stage_b_evidence_refs_seen":
                sum(1 for _ in _iter_evidence_refs_stage_b(minimal_brief)),
            "augmented_quote_count": len(augmented),
        },
    }


# ----------------------------------------------------------------------------
# Evidence Filter checker (W2a) — provenance + shape + cap + slug. No content
# scan; no lexicon/regex/boundary heuristics.
# ----------------------------------------------------------------------------
def _check_caps_evidence_pack(pack: dict, fails: list[str]) -> None:
    """W2-simple ContinuityBrief caps: per-list length + per-string
    length. Over-cap is FAIL only — no truncation."""
    list_keys = (
        "must_stay_consistent", "must_not_contradict",
        "allowed_creative_freedom", "state_change_rules",
        "unknowns_left_to_art_direction",
    )
    for key in list_keys:
        items = pack.get(key) or []
        if isinstance(items, list) and len(items) > CONTINUITY_BRIEF_LIST_MAX:
            fails.append(
                f"evidence_filter:cap_exceeded:{key}_items:"
                f"{len(items)}>{CONTINUITY_BRIEF_LIST_MAX}"
            )
    shot_checks = pack.get("shot_conflict_checks") or []
    if (isinstance(shot_checks, list)
            and len(shot_checks) > CONTINUITY_BRIEF_SHOT_CHECKS_MAX):
        fails.append(
            f"evidence_filter:cap_exceeded:shot_conflict_checks_items:"
            f"{len(shot_checks)}>{CONTINUITY_BRIEF_SHOT_CHECKS_MAX}"
        )
    # Per-string caps walk the rule items.
    def _cap_str(label: str, val: Any, limit: int) -> None:
        if isinstance(val, str) and len(val) > limit:
            fails.append(
                f"evidence_filter:cap_exceeded:{label}:{len(val)}>{limit}"
            )
    for key in ("must_stay_consistent", "must_not_contradict",
                "allowed_creative_freedom"):
        for it in (pack.get(key) or []):
            if not isinstance(it, dict):
                continue
            _cap_str(f"{key}_rule_id", it.get("rule_id"),
                     CONTINUITY_BRIEF_RULE_ID_CHARS)
            for f in ("statement", "do_not_introduce", "free_to_choose"):
                _cap_str(f"{key}_{f}", it.get(f),
                         CONTINUITY_BRIEF_STATEMENT_CHARS)
            for f in ("why_consistent", "why", "guidance"):
                _cap_str(f"{key}_{f}", it.get(f),
                         CONTINUITY_BRIEF_WHY_CHARS)
    for it in (pack.get("state_change_rules") or []):
        if not isinstance(it, dict):
            continue
        _cap_str("state_change_rules_rule_id", it.get("rule_id"),
                 CONTINUITY_BRIEF_RULE_ID_CHARS)
        _cap_str("state_change_rules_change_kind",
                 it.get("change_kind"), CONTINUITY_BRIEF_STATEMENT_CHARS)
        _cap_str("state_change_rules_when_applies",
                 it.get("when_applies"), CONTINUITY_BRIEF_STATEMENT_CHARS)
        for list_field in ("what_changes", "what_stays"):
            arr = it.get(list_field) or []
            if isinstance(arr, list):
                for entry in arr:
                    _cap_str(
                        f"state_change_rules_{list_field}_item",
                        entry, CONTINUITY_BRIEF_LIST_ITEM_CHARS,
                    )
    for it in (pack.get("shot_conflict_checks") or []):
        if not isinstance(it, dict):
            continue
        for list_field in ("must_support", "avoid_contradiction"):
            arr = it.get(list_field) or []
            if isinstance(arr, list):
                for entry in arr:
                    _cap_str(
                        f"shot_conflict_checks_{list_field}_item",
                        entry, CONTINUITY_BRIEF_LIST_ITEM_CHARS,
                    )
    for it in (pack.get("unknowns_left_to_art_direction") or []):
        if not isinstance(it, dict):
            continue
        _cap_str("unknowns_area", it.get("area"),
                 CONTINUITY_BRIEF_STATEMENT_CHARS)
        _cap_str("unknowns_note", it.get("note"),
                 CONTINUITY_BRIEF_LIST_ITEM_CHARS)
    cov = pack.get("coverage_notes") or []
    if isinstance(cov, list):
        if len(cov) > EVIDENCE_PACK_COVERAGE_NOTES_MAX:
            fails.append(
                f"evidence_filter:cap_exceeded:coverage_notes_items:"
                f"{len(cov)}>{EVIDENCE_PACK_COVERAGE_NOTES_MAX}"
            )
        for cn in cov:
            if not isinstance(cn, dict):
                continue
            note = cn.get("note") or ""
            if isinstance(note, str) and len(note) > EVIDENCE_PACK_COVERAGE_NOTE_CHARS:
                fails.append(
                    f"evidence_filter:cap_exceeded:coverage_note:"
                    f"{len(note)}>{EVIDENCE_PACK_COVERAGE_NOTE_CHARS}"
                )
            conseq = cn.get("consequence_for_background") or ""
            if isinstance(conseq, str) and len(conseq) > EVIDENCE_PACK_COVERAGE_CONSEQUENCE_CHARS:
                fails.append(
                    f"evidence_filter:cap_exceeded:coverage_consequence:"
                    f"{len(conseq)}>{EVIDENCE_PACK_COVERAGE_CONSEQUENCE_CHARS}"
                )
    rej = pack.get("rejected_or_irrelevant_summary") or []
    if isinstance(rej, list):
        if len(rej) > EVIDENCE_PACK_REJECTED_MAX:
            fails.append(
                f"evidence_filter:cap_exceeded:rejected_summary_items:"
                f"{len(rej)}>{EVIDENCE_PACK_REJECTED_MAX}"
            )
        for rn in rej:
            if not isinstance(rn, dict):
                continue
            sn = rn.get("short_note") or ""
            if isinstance(sn, str) and len(sn) > EVIDENCE_PACK_REJECTED_NOTE_CHARS:
                fails.append(
                    f"evidence_filter:cap_exceeded:rejected_short_note:"
                    f"{len(sn)}>{EVIDENCE_PACK_REJECTED_NOTE_CHARS}"
                )
            wn = rn.get("why_not_pass_down") or ""
            if isinstance(wn, str) and len(wn) > EVIDENCE_PACK_REJECTED_WHY_CHARS:
                fails.append(
                    f"evidence_filter:cap_exceeded:rejected_why_not_pass_down:"
                    f"{len(wn)}>{EVIDENCE_PACK_REJECTED_WHY_CHARS}"
                )


RULE_SECTIONS_WITH_EVIDENCE = (
    "must_stay_consistent", "must_not_contradict",
    "allowed_creative_freedom", "state_change_rules",
    "shot_conflict_checks",
)


def _iter_brief_rule_evidence_refs(pack: dict):
    """Iterate (section, idx, evidence_ref) triples across rule
    sections that carry evidence_refs. allowed_creative_freedom may
    have an empty list; we still iterate (no-op) so the caller's loop
    body never runs on a missing key."""
    for section in RULE_SECTIONS_WITH_EVIDENCE:
        for idx, rule in enumerate(pack.get(section) or []):
            if not isinstance(rule, dict):
                continue
            for er in (rule.get("evidence_refs") or []):
                yield (section, idx, er)


def _check_brief_rule_ids(pack: dict, fails: list[str]) -> None:
    seen: dict[str, set[str]] = {}
    for section in ("must_stay_consistent", "must_not_contradict",
                     "allowed_creative_freedom", "state_change_rules"):
        seen.setdefault(section, set())
        for idx, rule in enumerate(pack.get(section) or []):
            if not isinstance(rule, dict):
                fails.append(
                    f"evidence_filter:{section}_item_not_dict:{idx}"
                )
                continue
            rid = rule.get("rule_id")
            if not isinstance(rid, str) or not _slug_is_valid(rid):
                fails.append(
                    f"evidence_filter:rule_id_not_slug:{section}:{idx}:"
                    f"{rid!r}"
                )
            elif rid in seen[section]:
                fails.append(
                    f"evidence_filter:rule_id_duplicate:{section}:{rid}"
                )
            else:
                seen[section].add(rid)


def _check_common_place_identity(pack: dict, fails: list[str],
                                   source_text_by_ref: dict,
                                   augmented: list[dict]) -> int:
    cpi = pack.get("common_place_identity")
    er_count = 0
    if not isinstance(cpi, dict):
        fails.append(
            "evidence_filter:common_place_identity_missing_or_not_dict"
        )
        return er_count
    for k in ("summary", "why_this_identity", "confidence_band",
               "evidence_refs"):
        if k not in cpi:
            fails.append(
                f"evidence_filter:common_place_identity_missing_field:{k}"
            )
    cb = cpi.get("confidence_band")
    if cb not in CONFIDENCE_BANDS:
        fails.append(
            f"evidence_filter:common_place_identity_confidence_invalid:"
            f"{cb!r}"
        )
    summary = cpi.get("summary") or ""
    if (isinstance(summary, str)
            and len(summary) > CONTINUITY_BRIEF_IDENTITY_SUMMARY_CHARS):
        fails.append(
            f"evidence_filter:cap_exceeded:identity_summary:"
            f"{len(summary)}>{CONTINUITY_BRIEF_IDENTITY_SUMMARY_CHARS}"
        )
    why = cpi.get("why_this_identity") or ""
    if (isinstance(why, str)
            and len(why) > CONTINUITY_BRIEF_IDENTITY_WHY_CHARS):
        fails.append(
            f"evidence_filter:cap_exceeded:identity_why:"
            f"{len(why)}>{CONTINUITY_BRIEF_IDENTITY_WHY_CHARS}"
        )
    refs = cpi.get("evidence_refs") or []
    if not isinstance(refs, list) or not refs:
        fails.append(
            "evidence_filter:common_place_identity_evidence_refs_empty"
        )
    else:
        for idx, er in enumerate(refs):
            er_count += 1
            if not isinstance(er, dict):
                fails.append(
                    f"evidence_filter:evidence_ref_not_dict:"
                    f"common_place_identity:{idx}"
                )
                continue
            sr = er.get("source_ref")
            if sr not in source_text_by_ref:
                fails.append(
                    f"evidence_filter:source_ref_unresolved:"
                    f"common_place_identity:{idx}:{sr!r}"
                )
                continue
            q = er.get("quote") or ""
            if not isinstance(q, str) or not q:
                fails.append(
                    f"evidence_filter:quote_missing:"
                    f"common_place_identity:{idx}"
                )
                continue
            pos = source_text_by_ref[sr].find(q)
            if pos < 0:
                fails.append(
                    f"evidence_filter:quote_not_in_source:"
                    f"common_place_identity:{idx}"
                )
            else:
                augmented.append({
                    "section": "common_place_identity",
                    "rule_index": idx,
                    "source_ref": sr,
                    "char_start": pos,
                    "char_end": pos + len(q),
                })
            # Codex review Q3: evidence_ref must NOT carry
            # confidence_band. Confidence sits at the rule level.
            if "confidence_band" in er:
                fails.append(
                    f"evidence_filter:evidence_ref_has_forbidden_"
                    f"confidence_band:common_place_identity:{idx}"
                )
    return er_count


def run_evidence_filter_validation(*, evidence_pack: dict,
                                    bundle: SourceBundle,
                                    schema: dict) -> dict:
    """W2-simple ContinuityBrief checker (Codex consultation APPROVED).

    Validates:
      - top-level required keys (including common_place_identity)
      - source_bundle_ref matches the bundle
      - common_place_identity: required fields + rule-level
        confidence_band + ≥1 evidence_ref (provenance only)
      - per-rule rule_id is a slug + unique within its section
      - rule-level confidence_band ∈ enum (all rule sections)
      - allowed_creative_freedom: `basis` enum, evidence_refs optional
      - shot_conflict_checks: shot_ref exact-equality with a
        selected_shot source_ref in the SourceBundle (if any)
      - every evidence_ref's source_ref resolves + quote is verbatim
      - evidence_ref MUST NOT carry confidence_band (it lives on the
        rule, not the provenance)
      - coverage_notes confidence_band ∈ enum
      - no forbidden top-level / item-level keys
      - all string + list caps

    NO semantic role classification, NO content scan.
    """
    fails: list[str] = []
    augmented: list[dict] = []
    for k in schema.get("required", []):
        if k not in evidence_pack:
            fails.append(f"evidence_filter:missing_top_level_key:{k}")
    sbr = evidence_pack.get("source_bundle_ref")
    if sbr and sbr != bundle.bundle_id:
        fails.append(
            f"evidence_filter:source_bundle_ref_mismatch:{sbr!r}!="
            f"{bundle.bundle_id!r}"
        )
    source_text_by_ref = {s.source_ref: s.text for s in bundle.sources}
    selected_shot_refs = {
        s.source_ref for s in bundle.sources if s.kind == "selected_shot"
    }
    er_count = _check_common_place_identity(
        evidence_pack, fails, source_text_by_ref, augmented,
    )
    _check_brief_rule_ids(evidence_pack, fails)
    # Rule-level confidence_band (per section).
    for section in RULE_SECTIONS_WITH_EVIDENCE:
        for idx, rule in enumerate(evidence_pack.get(section) or []):
            if not isinstance(rule, dict):
                continue
            cb = rule.get("confidence_band")
            if cb not in CONFIDENCE_BANDS:
                fails.append(
                    f"evidence_filter:rule_confidence_invalid:"
                    f"{section}:{idx}:{cb!r}"
                )
    # allowed_creative_freedom: basis enum check.
    for idx, rule in enumerate(
            evidence_pack.get("allowed_creative_freedom") or []):
        if not isinstance(rule, dict):
            continue
        basis = rule.get("basis")
        if basis not in CONTINUITY_BRIEF_ALLOWED_BASIS_ENUM:
            fails.append(
                f"evidence_filter:allowed_creative_freedom_basis_"
                f"invalid:{idx}:{basis!r}"
            )
    # shot_conflict_checks: shot_ref exact equality with a
    # selected_shot source_ref present in the SourceBundle, AND at
    # least one evidence_ref must cite that same shot_ref (Codex W2-
    # simple-b BLOCKING 1: episode-only refs cannot ground a per-shot
    # conflict check).
    for idx, rule in enumerate(
            evidence_pack.get("shot_conflict_checks") or []):
        if not isinstance(rule, dict):
            continue
        sref = rule.get("shot_ref")
        if not isinstance(sref, str) or not sref:
            fails.append(
                f"evidence_filter:shot_ref_missing:{idx}"
            )
            continue
        if selected_shot_refs and sref not in selected_shot_refs:
            fails.append(
                f"evidence_filter:shot_ref_not_in_selected_shots:"
                f"{idx}:{sref!r}"
            )
            continue
        ev_refs = rule.get("evidence_refs") or []
        if not any(
            isinstance(er, dict) and er.get("source_ref") == sref
            for er in ev_refs
        ):
            fails.append(
                f"evidence_filter:shot_conflict_check_missing_"
                f"matching_shot_evidence:{idx}:{sref!r}"
            )
    # Walk per-rule evidence_refs across all sections (excluding
    # common_place_identity, which was handled above).
    for section, idx, er in _iter_brief_rule_evidence_refs(evidence_pack):
        er_count += 1
        if not isinstance(er, dict):
            fails.append(
                f"evidence_filter:evidence_ref_not_dict:{section}:{idx}"
            )
            continue
        sr = er.get("source_ref")
        if sr not in source_text_by_ref:
            fails.append(
                f"evidence_filter:source_ref_unresolved:{section}:{idx}:"
                f"{sr!r}"
            )
            continue
        q = er.get("quote") or ""
        if not isinstance(q, str) or not q:
            fails.append(
                f"evidence_filter:quote_missing:{section}:{idx}"
            )
            continue
        pos = source_text_by_ref[sr].find(q)
        if pos < 0:
            fails.append(
                f"evidence_filter:quote_not_in_source:{section}:{idx}"
            )
        else:
            augmented.append({
                "section": section,
                "rule_index": idx,
                "source_ref": sr,
                "char_start": pos,
                "char_end": pos + len(q),
            })
        # evidence_ref must NOT carry confidence_band (Codex Q3).
        if "confidence_band" in er:
            fails.append(
                f"evidence_filter:evidence_ref_has_forbidden_"
                f"confidence_band:{section}:{idx}"
            )
    cov = evidence_pack.get("coverage_notes") or []
    if isinstance(cov, list):
        for idx, cn in enumerate(cov):
            if not isinstance(cn, dict):
                continue
            cb = cn.get("confidence_band")
            if cb not in CONFIDENCE_BANDS:
                fails.append(
                    f"evidence_filter:coverage_confidence_invalid:"
                    f"{idx}:{cb!r}"
                )
    _check_no_forbidden_keys(
        obj=evidence_pack, stage_label="evidence_filter",
        forbidden=schema.get("forbidden_top_level_or_item_fields", []),
        path="$", fails=fails,
    )
    _check_caps_evidence_pack(evidence_pack, fails)
    return {
        "executed": True,
        "passed": len(fails) == 0,
        "failed_checks": fails,
        "augmented_occurrences": augmented,
        "summary": {
            "common_place_identity_present":
                isinstance(evidence_pack.get("common_place_identity"),
                            dict),
            "must_stay_consistent_count":
                len(evidence_pack.get("must_stay_consistent") or []),
            "must_not_contradict_count":
                len(evidence_pack.get("must_not_contradict") or []),
            "allowed_creative_freedom_count":
                len(evidence_pack.get("allowed_creative_freedom") or []),
            "state_change_rules_count":
                len(evidence_pack.get("state_change_rules") or []),
            "shot_conflict_checks_count":
                len(evidence_pack.get("shot_conflict_checks") or []),
            "unknowns_count":
                len(
                    evidence_pack.get("unknowns_left_to_art_direction")
                    or []
                ),
            "coverage_notes_count": (
                len(cov) if isinstance(cov, list) else 0
            ),
            "evidence_ref_count": er_count,
            "augmented_quote_count": len(augmented),
        },
    }


def planned_evidence_filter_report() -> dict:
    return {
        "executed": False,
        "planned_checks": [
            "evidence_filter:schema_required_fields (5 top-level)",
            "evidence_filter:evidence_item_required_fields "
            "(evidence_id, source_ref, quote, why_background_relevant, "
            "use_for, confidence_band, pass_down_hint)",
            "evidence_filter:source_bundle_ref_matches_bundle",
            "evidence_filter:evidence_id_slug_only "
            "(ASCII alnum + _/-, char loop)",
            "evidence_filter:source_ref_resolvable "
            "(matches a SourceBundle source)",
            "evidence_filter:quote_exact_containment "
            "(source.text contains exact quote)",
            "evidence_filter:use_for_enum "
            "(world_context | place_identity | "
            "spatial_scale_or_layout_feel | state_variation | "
            "visual_default_or_avoid)",
            "evidence_filter:confidence_band_enum "
            "(trusted | plausible | weak | unknown)",
            "evidence_filter:no_forbidden_fields "
            "(topology + count + door/window/furniture/object/inventory "
            "keys; JSON key-name detection only)",
            "evidence_filter:minimality_caps "
            "(evidence_items ≤ 40; why ≤ 180 chars; "
            "pass_down_hint ≤ 160 chars; coverage_notes ≤ 8; "
            "rejected_or_irrelevant_summary ≤ 8; over-cap is FAIL — "
            "no auto-truncate, original pack preserved verbatim)",
        ],
        "note": (
            "Evidence Filter checker is the gateway provenance layer. It "
            "validates the LLM-emitted pack's shape, source_ref + quote "
            "containment, enum membership, and caps. It does NOT classify "
            "meaning or scan pack/source content for forbidden words. "
            "Semantic adequacy (did the filter actually capture the most "
            "useful evidence?) is deferred to human / LLM review."
        ),
    }


def placeholder_evidence_pack(bundle: SourceBundle) -> dict:
    return {
        "placeholder": True,
        "stage": "evidence_filter",
        "source_bundle_ref": bundle.bundle_id,
        "schema_version": EVIDENCE_PACK_SCHEMA_VERSION,
        "reason": "dry-run only; --generate default off",
    }


def route_evidence_filter_result(*, filter_result: dict,
                                  bundle: SourceBundle,
                                  schema: dict,
                                  correction_call: Optional[Any] = None,
                                  model: Optional[str] = None) -> dict:
    """W2a-b BLOCKING 1 + W2a-d correction pass.

    Separate "LLM call succeeded" from "EvidencePack is valid". If the
    initial validation fails *and* the failure includes any
    `quote_not_in_source` entry, invoke `correction_call` exactly ONCE
    with the invalid pack + failed_checks; re-validate; promote on
    success or quarantine on failure. If `correction_call` is None or
    failure has no `quote_not_in_source` entries, fail-closed as before
    (W2a-b semantics, no correction).

    Returns:
        {
            "downstream_pack": dict,           # placeholder OR valid pack
            "quarantined_pack": dict | None,
            "evidence_filter_report": dict,    # includes correction_* meta
            "run_status": "succeeded"
                         | "validation_failed"
                         | "generate_failed",
            "exit_code": 0 if succeeded else 1,
        }
    """
    if filter_result.get("status") == "succeeded":
        pack = filter_result.get("evidence_pack") or {}
        initial_attempts = filter_result.get("attempts") or 0
        report = run_evidence_filter_validation(
            evidence_pack=pack, bundle=bundle, schema=schema,
        )
        report["generate_status"] = filter_result.get("status")
        report["generate_attempts"] = initial_attempts
        report["generate_model"] = filter_result.get("model")
        report["evidence_filter_attempts"] = initial_attempts
        report.setdefault("correction_attempted", False)
        report.setdefault("correction_status", "not_attempted")
        report.setdefault("correction_model", None)
        report.setdefault("correction_failed_checks", [])
        if report.get("passed"):
            report["run_status"] = "succeeded"
            return {
                "downstream_pack": pack,
                "quarantined_pack": None,
                "evidence_filter_report": report,
                "run_status": "succeeded",
                "exit_code": 0,
            }
        initial_failed = list(report.get("failed_checks") or [])
        has_quote_fail = any(
            "quote_not_in_source" in fc for fc in initial_failed
        )
        if correction_call is not None and has_quote_fail:
            correction_model = model or filter_result.get("model")
            report["correction_attempted"] = True
            report["correction_model"] = correction_model
            report["initial_failed_checks"] = initial_failed
            try:
                corrected_pack = correction_call(
                    bundle=bundle, schema=schema,
                    model=correction_model,
                    invalid_pack=pack,
                    failed_checks=initial_failed,
                )
            except Exception as exc:  # noqa: BLE001
                report["correction_status"] = "failed"
                report["correction_error"] = (
                    f"{type(exc).__name__}: {exc}"
                )
                report["evidence_filter_attempts"] = initial_attempts + 1
                placeholder = {
                    "placeholder": True,
                    "stage": "evidence_filter",
                    "source_bundle_ref": bundle.bundle_id,
                    "schema_version": EVIDENCE_PACK_SCHEMA_VERSION,
                    "reason": (
                        f"evidence_filter validation failed "
                        f"({len(initial_failed)} fail(s)); correction "
                        f"pass also failed ({report['correction_error']})."
                    ),
                    "model": filter_result.get("model"),
                }
                report["run_status"] = "validation_failed"
                return {
                    "downstream_pack": placeholder,
                    "quarantined_pack": pack,
                    "evidence_filter_report": report,
                    "run_status": "validation_failed",
                    "exit_code": 1,
                }
            correction_validation = run_evidence_filter_validation(
                evidence_pack=corrected_pack, bundle=bundle, schema=schema,
            )
            report["evidence_filter_attempts"] = initial_attempts + 1
            if correction_validation.get("passed"):
                # Promote corrected pack. Merge correction summary into the
                # report but keep the original metadata for audit.
                report["correction_status"] = "succeeded"
                report["correction_failed_checks"] = []
                report["passed"] = True
                report["failed_checks"] = []
                report["augmented_occurrences"] = (
                    correction_validation.get("augmented_occurrences") or []
                )
                report["summary"] = correction_validation.get("summary") or {}
                report["run_status"] = "succeeded"
                return {
                    "downstream_pack": corrected_pack,
                    "quarantined_pack": None,
                    "evidence_filter_report": report,
                    "run_status": "succeeded",
                    "exit_code": 0,
                }
            # Correction returned a pack that still fails validation.
            report["correction_status"] = "validation_failed"
            report["correction_failed_checks"] = list(
                correction_validation.get("failed_checks") or []
            )
            # Keep the original failed_checks visible at the top level
            # for the HTML cockpit; the correction round's failures live
            # under correction_failed_checks.
            placeholder = {
                "placeholder": True,
                "stage": "evidence_filter",
                "source_bundle_ref": bundle.bundle_id,
                "schema_version": EVIDENCE_PACK_SCHEMA_VERSION,
                "reason": (
                    f"evidence_filter validation failed "
                    f"({len(initial_failed)} initial fail(s)); correction "
                    f"pass also failed "
                    f"({len(report['correction_failed_checks'])} "
                    f"fail(s)); latest pack quarantined."
                ),
                "model": filter_result.get("model"),
            }
            report["run_status"] = "validation_failed"
            return {
                "downstream_pack": placeholder,
                "quarantined_pack": corrected_pack,
                "evidence_filter_report": report,
                "run_status": "validation_failed",
                "exit_code": 1,
            }
        # No correction (either correction_call is None, or fail set has
        # no quote_not_in_source — conservative fail-closed).
        report["correction_attempted"] = False
        report["correction_status"] = "skipped"
        placeholder = {
            "placeholder": True,
            "stage": "evidence_filter",
            "source_bundle_ref": bundle.bundle_id,
            "schema_version": EVIDENCE_PACK_SCHEMA_VERSION,
            "reason": (
                f"evidence_filter validation failed "
                f"({len(initial_failed)} fail(s)); original pack "
                f"quarantined, downstream prompts use this placeholder."
            ),
            "model": filter_result.get("model"),
        }
        report["run_status"] = "validation_failed"
        return {
            "downstream_pack": placeholder,
            "quarantined_pack": pack,
            "evidence_filter_report": report,
            "run_status": "validation_failed",
            "exit_code": 1,
        }
    # filter_result["status"] != "succeeded" — generate failure.
    placeholder = {
        "placeholder": True,
        "stage": "evidence_filter",
        "source_bundle_ref": bundle.bundle_id,
        "schema_version": EVIDENCE_PACK_SCHEMA_VERSION,
        "reason": (
            f"evidence_filter_generate failed after "
            f"{filter_result.get('attempts')} attempt(s)"
        ),
        "error": filter_result.get("error", ""),
        "model": filter_result.get("model"),
    }
    report = {
        "executed": False,
        "passed": False,
        "failed_checks": [
            f"evidence_filter:generate_failed:"
            f"{filter_result.get('error', '')}",
        ],
        "augmented_occurrences": [],
        "generate_status": filter_result.get("status"),
        "generate_attempts": filter_result.get("attempts"),
        "generate_model": filter_result.get("model"),
        "run_status": "generate_failed",
        "summary": {
            "evidence_items_count": 0,
            "coverage_notes_count": 0,
            "augmented_quote_count": 0,
        },
        "correction_attempted": False,
        "correction_status": "not_attempted",
        "correction_model": None,
        "correction_failed_checks": [],
        "evidence_filter_attempts": filter_result.get("attempts") or 0,
        "note": (
            "Evidence Filter generation failed; downstream stages use "
            "a placeholder pack so the run still emits auditable "
            "artifacts."
        ),
    }
    return {
        "downstream_pack": placeholder,
        "quarantined_pack": None,
        "evidence_filter_report": report,
        "run_status": "generate_failed",
        "exit_code": 1,
    }


def run_evidence_filter_generate(*, bundle: SourceBundle, schema: dict,
                                  model: str, llm_call,
                                  max_retries: int = 1) -> dict:
    """Call the Evidence Filter LLM (callable injected for testability).

    Total attempts = 1 + max_retries (initial + retries). Never raises;
    returns a status dict. Caller decides what to write to disk.
    """
    if max_retries < 0:
        max_retries = 0
    attempts = 0
    last_error = ""
    while attempts <= max_retries:
        attempts += 1
        try:
            pack = llm_call(bundle=bundle, schema=schema, model=model)
        except Exception as e:  # noqa: BLE001
            last_error = f"{type(e).__name__}: {e}"
            continue
        return {
            "status": "succeeded",
            "attempts": attempts,
            "evidence_pack": pack,
            "model": model,
            "error": "",
        }
    return {
        "status": "failed",
        "attempts": attempts,
        "evidence_pack": None,
        "model": model,
        "error": last_error,
    }


def _default_evidence_filter_llm_call(*, bundle: SourceBundle,
                                       schema: dict, model: str) -> dict:
    """Default LLM call for the Evidence Filter — invoked only when
    `--generate` is on. Uses LiteLLM (the backend's existing gateway)
    so the script does not depend on a provider SDK installed only in
    `backend/app`. Raises on any failure so the retry/fail-closed
    wrapper in run_evidence_filter_generate can record an explicit
    failed status.
    """
    spec = build_sample_fixture_l05_spec()
    system_txt = build_prompt_evidence_filter_system(spec)
    user_txt = build_prompt_evidence_filter_user(
        spec=spec, bundle=bundle, schema=schema,
    )
    if not isinstance(model, str) or not model.strip():
        raise RuntimeError(f"unsupported_model: {model!r}")
    if model.lower().startswith("gemini") and not os.environ.get(
        "GEMINI_API_KEY"
    ) and not os.environ.get("GOOGLE_API_KEY"):
        raise RuntimeError(
            "missing GEMINI_API_KEY / GOOGLE_API_KEY env var"
        )
    routed_model = model
    if not routed_model.startswith("gemini/") and routed_model.lower().startswith("gemini"):
        routed_model = "gemini/" + routed_model
    import litellm  # noqa: PLC0415
    response = litellm.completion(
        model=routed_model,
        messages=[
            {"role": "system", "content": system_txt},
            {"role": "user", "content": user_txt},
        ],
        response_format={"type": "json_object"},
    )
    choices = getattr(response, "choices", None) or []
    if not choices:
        raise RuntimeError("empty response: no choices from litellm")
    msg = getattr(choices[0], "message", None) or {}
    content = (
        getattr(msg, "content", None)
        if not isinstance(msg, dict)
        else msg.get("content")
    )
    if not content or not content.strip():
        raise RuntimeError("empty response: blank message content")
    body = content.strip()
    if body.startswith("```"):
        first_nl = body.find("\n")
        if first_nl >= 0:
            body = body[first_nl + 1:]
        if body.endswith("```"):
            body = body[: -len("```")]
    try:
        return json.loads(body)
    except json.JSONDecodeError as je:
        raise RuntimeError(
            f"non-JSON response: {je}; first 200 chars={body[:200]!r}"
        )


def _default_evidence_filter_correction_llm_call(
        *, bundle: SourceBundle, schema: dict, model: str,
        invalid_pack: dict, failed_checks: list[str]) -> dict:
    """W2a-d default correction LLM call. Same provider routing as the
    initial call; raises on any failure so the routing wrapper can record
    correction_status='failed'. NO source text truncation."""
    spec = build_sample_fixture_l05_spec()
    system_txt = build_prompt_evidence_filter_correction_system(spec)
    user_txt = build_prompt_evidence_filter_correction_user(
        spec=spec, bundle=bundle, schema=schema,
        invalid_pack=invalid_pack, failed_checks=failed_checks,
    )
    if not isinstance(model, str) or not model.strip():
        raise RuntimeError(f"unsupported_model: {model!r}")
    if model.lower().startswith("gemini") and not os.environ.get(
        "GEMINI_API_KEY"
    ) and not os.environ.get("GOOGLE_API_KEY"):
        raise RuntimeError(
            "missing GEMINI_API_KEY / GOOGLE_API_KEY env var"
        )
    routed_model = model
    if (not routed_model.startswith("gemini/")
            and routed_model.lower().startswith("gemini")):
        routed_model = "gemini/" + routed_model
    import litellm  # noqa: PLC0415
    response = litellm.completion(
        model=routed_model,
        messages=[
            {"role": "system", "content": system_txt},
            {"role": "user", "content": user_txt},
        ],
        response_format={"type": "json_object"},
    )
    choices = getattr(response, "choices", None) or []
    if not choices:
        raise RuntimeError("empty correction response: no choices")
    msg = getattr(choices[0], "message", None) or {}
    content = (
        getattr(msg, "content", None)
        if not isinstance(msg, dict)
        else msg.get("content")
    )
    if not content or not content.strip():
        raise RuntimeError("empty correction response: blank content")
    body = content.strip()
    if body.startswith("```"):
        first_nl = body.find("\n")
        if first_nl >= 0:
            body = body[first_nl + 1:]
        if body.endswith("```"):
            body = body[: -len("```")]
    try:
        return json.loads(body)
    except json.JSONDecodeError as je:
        raise RuntimeError(
            f"non-JSON correction response: {je}; first 200 chars="
            f"{body[:200]!r}"
        )


def planned_checker_report() -> dict:
    return {
        "executed": False,
        "planned_checks": [
            "1. stage_a_schema_required_fields (4 top-level + 8 wbb subfields)",
            "2. stage_b_schema_required_fields (11 top-level + per-item required)",
            "3. stage_a_evidence_refs_non_empty (world_background_brief)",
            "4. stage_b_evidence_refs_non_empty (place_identity + 4 list groups)",
            "5. source_ref_resolvable (both stages, matches SourceBundle)",
            "6. quote_exact_containment (both stages, source.text contains exact quote)",
            "7. confidence_enum (both stages, trusted/plausible/weak/unknown)",
            "8. no_forbidden_fields (topology fields + 'count' JSON key + Stage B embedded world dict)",
            "9. minimality_caps (array length + string length caps; equal-to-cap PASS, over-cap FAIL — NO auto-truncate/slice/drop/chunk/auto-shorten; original brief content preserved verbatim)",
            "10. stage_b_input_contract (structural: world keys exactly {world_brief_ref, world_hints_for_background}; no embedded full Stage A dict; no irrelevant_or_do_not_pass_down)",
        ],
        "note": (
            "checker does NOT classify semantic meaning, does NOT scan "
            "brief content for forbidden words, and does NOT apply any "
            "lexicon/regex/boundary heuristic. checker does NOT truncate, "
            "slice, drop, chunk, or auto-shorten any brief field or source "
            "text — cap violations are recorded as fail entries; original "
            "values are preserved verbatim. Semantic checks (LLM passing "
            "plot/lore into hints despite Stage A irrelevant list) are "
            "deferred to W2 LLM/human review."
        ),
    }


# ============================================================================
# Generate fail-closed (W0b IMPORTANT 1)
# ============================================================================
class SourceTooLargeForModel(RuntimeError):
    """plan §0 #8 / §11 — raised when --generate is on but budget exceeds spec limit."""


def check_char_budget_fail_closed(*, bundle: SourceBundle,
                                   spec: SampleFixtureSpec) -> None:
    if bundle.char_budget_estimate > spec.model_context_char_limit:
        raise SourceTooLargeForModel(
            f"source_too_large_for_model: "
            f"bundle.char_budget_estimate={bundle.char_budget_estimate} > "
            f"spec.model_context_char_limit={spec.model_context_char_limit}"
        )


# ============================================================================
# HTML render
# ============================================================================
def _esc(value: Any) -> str:
    return html_lib.escape(str(value), quote=True)


def render_html(*, bundle: SourceBundle, world_schema: dict,
                 minimal_schema: dict, evidence_pack_schema: dict,
                 evidence_pack: dict, evidence_filter_report: dict,
                 prompt_evidence_filter_system: str,
                 prompt_evidence_filter_user: str,
                 prompt_world_system: str,
                 prompt_world_user: str, prompt_spatial_system: str,
                 prompt_spatial_user: str, validation_report: dict,
                 run_meta: dict, include_diagnostic: bool) -> str:
    parts: list[str] = []
    parts.append(
        "<!doctype html><html lang='ko'><head><meta charset='utf-8'>"
        "<title>Background Semantic Extractor — W2-simple "
        "(BackgroundContinuityBrief)</title>"
        "<style>body{font-family:-apple-system,BlinkMacSystemFont,system-ui,"
        "sans-serif;max-width:1180px;margin:24px auto;padding:0 16px;"
        "color:#222} h1,h2{border-bottom:1px solid #ddd;padding-bottom:4px}"
        " table{border-collapse:collapse;width:100%;margin:8px 0}"
        " th,td{border:1px solid #ddd;padding:4px 8px;font-size:13px;"
        "vertical-align:top}"
        " code{background:#f6f8fa;padding:1px 4px;border-radius:3px}"
        " pre{background:#f6f8fa;padding:8px;border-radius:4px;overflow-x:auto}"
        " .badge{display:inline-block;padding:2px 8px;border-radius:10px;"
        "background:#eef;font-size:11px;margin-right:4px}"
        " .ok{color:#0a0}.bad{color:#c00}.warn{color:#a60}"
        "</style></head><body>"
    )
    parts.append(
        f"<h1>Background Semantic Extractor — generic experiment, minimal "
        f"2-stage brief (sample fixture: <code>{_esc(bundle.fixture_id)}</code>)</h1>"
    )
    rs = (run_meta.get("run_status") or "dry_run")
    correction_attempted = bool(run_meta.get("correction_attempted"))
    correction_status = run_meta.get("correction_status") or "not_attempted"
    if rs == "succeeded" and correction_attempted:
        badge_html = (
            "<span class='badge' style='background:#cfc;color:#070'>"
            "succeeded (filter validated after correction)</span>"
        )
        initial_fails = run_meta.get("initial_failed_checks") or []
        quote_fail_count = sum(
            1 for f in initial_fails if "quote_not_in_source" in f
        )
        other_fail_count = len(initial_fails) - quote_fail_count
        warning_html = (
            "<p style='background:#efe;border:1px solid #6c6;"
            "padding:8px;color:#070'>"
            "<strong>Correction succeeded.</strong> "
            f"initial quote failures corrected: <code>{quote_fail_count}"
            f"</code>"
            + (f"; other initial failures: <code>{other_fail_count}</code>"
               if other_fail_count else "")
            + ". The final pack passes deterministic validation; "
            "downstream world / spatial prompts consume this pack.</p>"
        )
    elif rs == "validation_failed" and correction_attempted:
        badge_html = (
            "<span class='badge' style='background:#fcc;color:#900'>"
            "validation_failed (correction "
            f"{_esc(correction_status)} — pack quarantined)</span>"
        )
        warning_html = (
            "<p style='background:#fee;border:1px solid #c00;"
            "padding:8px;color:#900'>"
            "<strong>⚠ filter validation failed, correction pass also "
            "did not produce a valid pack.</strong> The initial Evidence "
            "Filter LLM responded but its pack failed validation; the "
            "correction pass (with the failed_checks fed back to the "
            "LLM) also did not pass. The latest pack has been "
            "quarantined to "
            "<code>background_evidence_pack_quarantined.json</code>; "
            "downstream world/spatial prompts use a placeholder. Treat "
            "this run as a diagnostic artifact, not an approved "
            "EvidencePack.</p>"
        )
    else:
        badge_map = {
            "dry_run": (
                "<span class='badge'>dry-run (no LLM call)</span>",
                "",
            ),
            "succeeded": (
                "<span class='badge' style='background:#cfc;color:#070'>"
                "succeeded (filter validated)</span>",
                "",
            ),
            "validation_failed": (
                "<span class='badge' style='background:#fcc;color:#900'>"
                "validation_failed (fail-closed, pack quarantined)"
                "</span>",
                (
                    "<p style='background:#fee;border:1px solid #c00;"
                    "padding:8px;color:#900'>"
                    "<strong>⚠ filter validation failed.</strong> The "
                    "Evidence Filter LLM responded, but the resulting "
                    "BackgroundEvidencePack failed deterministic "
                    "validation. The invalid pack has been quarantined "
                    "to <code>background_evidence_pack_quarantined.json"
                    "</code>; downstream world/spatial prompts use a "
                    "placeholder. Treat this run as a diagnostic "
                    "artifact, not an approved EvidencePack.</p>"
                ),
            ),
            "generate_failed": (
                "<span class='badge' style='background:#fcc;color:#900'>"
                "generate_failed (LLM call did not return a usable "
                "response)</span>",
                (
                    "<p style='background:#fee;border:1px solid #c00;"
                    "padding:8px;color:#900'>"
                    "<strong>⚠ Evidence Filter generate failed.</strong> "
                    "The LLM call raised an error (see "
                    "<code>filter_validation_report.json</code>). "
                    "Downstream stages used a placeholder.</p>"
                ),
            ),
        }
        badge_html, warning_html = badge_map.get(rs, badge_map["dry_run"])
    parts.append(
        f"<p>{badge_html}"
        f" run_id=<code>{_esc(run_meta.get('run_id'))}</code>"
        f" / plan_version=<code>{_esc(run_meta.get('plan_version'))}</code>"
        f" / model=<code>{_esc(run_meta.get('model'))}</code>"
        f" / run_status=<code>{_esc(rs)}</code>"
        f" / exit_code=<code>{_esc(run_meta.get('exit_code'))}</code>"
        f" / <strong>Actual downstream input = "
        f"BackgroundContinuityBrief</strong></p>"
    )
    if warning_html:
        parts.append(warning_html)
    parts.append(
        "<h2>§0. Why minimal + why 2-stage + why an Evidence Filter</h2>"
        "<p>Minimal means we extract the smallest evidence-backed brief "
        "that keeps T2I/I2I background generation honest. Over-extraction "
        "of door/window/furniture inventory tends to hurt downstream shot "
        "generation, so the schema avoids object counting and rigid "
        "inventory.</p>"
        "<p>2-stage means Stage A compresses the world grounding (era, "
        "geography, technology, social/economic tone, genre mood) and "
        "explicitly drops plot/lore that does not change visual background. "
        "Stage B uses only the short world hints from Stage A, never the "
        "full Stage A dict — enforced by the structural input-contract "
        "check.</p>"
        "<p><strong>Full SourceBundle was read only by the Evidence Filter. "
        "Downstream background briefs use only filtered evidence.</strong> "
        "The Filter is the single gateway over the raw SourceBundle; the "
        "downstream world / spatial prompt builders only receive the "
        "filtered BackgroundEvidencePack, never the bundle.</p>"
    )
    if include_diagnostic:
        parts.append(
            "<p><span class='warn'>⚠ deprecated diagnostic input</span> — "
            "this run included existing literal/pattern-era artifacts as "
            "<code>existing_artifact</code> sources. Treat with caution.</p>"
        )
    # ── §1. BackgroundContinuityBrief (downstream input) ──
    parts.append("<h2>§1. BackgroundContinuityBrief — downstream input</h2>")
    parts.append(
        "<p>This is the actual input the downstream world / spatial "
        "stages consume. Constraint statements (not image commands). "
        "The raw SourceBundle (read only by the Evidence Filter LLM) "
        "is collapsed to the bottom of this page.</p>"
    )
    if evidence_pack.get("placeholder"):
        reason = (evidence_pack.get("reason") or "").lower()
        if rs == "validation_failed":
            parts.append(
                "<p><span class='badge' style='background:#fcc;color:#900'>"
                "placeholder (invalid brief quarantined)</span> "
                "The LLM did respond, but the resulting brief failed "
                "validation; the original is in "
                "<code>background_evidence_pack_quarantined.json</code> "
                "and downstream prompts use this placeholder.</p>"
            )
        elif rs == "generate_failed":
            parts.append(
                "<p><span class='badge' style='background:#fcc;color:#900'>"
                "placeholder (generate failed)</span> "
                "The LLM call did not return a usable response; "
                "downstream prompts use this placeholder.</p>"
            )
        elif "validation failed" in reason or "quarantined" in reason:
            parts.append(
                "<p><span class='badge'>placeholder brief</span> "
                f"{_esc(evidence_pack.get('reason', ''))}</p>"
            )
        else:
            parts.append(
                "<p><span class='badge'>placeholder brief</span> "
                "Filter LLM not called in dry-run; downstream prompts use "
                "this placeholder.</p>"
            )
    else:
        # §1.0 Brief summary counts.
        counts = {
            "common_place_identity":
                1 if isinstance(
                    evidence_pack.get("common_place_identity"), dict,
                ) else 0,
            "must_stay_consistent": len(
                evidence_pack.get("must_stay_consistent") or []),
            "must_not_contradict": len(
                evidence_pack.get("must_not_contradict") or []),
            "allowed_creative_freedom": len(
                evidence_pack.get("allowed_creative_freedom") or []),
            "state_change_rules": len(
                evidence_pack.get("state_change_rules") or []),
            "shot_conflict_checks": len(
                evidence_pack.get("shot_conflict_checks") or []),
            "unknowns": len(
                evidence_pack.get("unknowns_left_to_art_direction") or []),
            "coverage_notes": len(
                evidence_pack.get("coverage_notes") or []),
            "rejected": len(
                evidence_pack.get("rejected_or_irrelevant_summary") or []),
        }
        count_strs = [
            f"<code>{_esc(k)}</code>: <strong>{v}</strong>"
            for k, v in counts.items() if v
        ]
        parts.append(
            "<p>Actual downstream input = BackgroundContinuityBrief. "
            "Section counts — " + " / ".join(count_strs) + "</p>"
        )
        # §1.1 common_place_identity.
        cpi = evidence_pack.get("common_place_identity")
        if isinstance(cpi, dict):
            parts.append(
                "<h3>§1.1 common_place_identity "
                "(shared baseline every shot must respect)</h3>"
            )
            parts.append(
                f"<p><strong>{_esc(cpi.get('summary', ''))}</strong> "
                f"<em>({_esc(cpi.get('confidence_band', ''))})</em></p>"
            )
            parts.append(
                f"<p>why: {_esc(cpi.get('why_this_identity', ''))}</p>"
            )
            er_list = cpi.get("evidence_refs") or []
            if er_list:
                parts.append("<ul>")
                for er in er_list:
                    if not isinstance(er, dict):
                        continue
                    parts.append(
                        f"<li><code>{_esc(er.get('source_ref', ''))}"
                        f"</code>: <em>{_esc(er.get('quote', ''))}</em>"
                        "</li>"
                    )
                parts.append("</ul>")

        def _render_rule_section(
                section_key: str, label: str,
                primary_field: str, secondary_field: str) -> None:
            rules = evidence_pack.get(section_key) or []
            if not rules:
                return
            parts.append(
                f"<h3>{_esc(label)} "
                f"(count={len(rules)})</h3>"
            )
            parts.append(
                "<table><thead><tr>"
                "<th>rule_id</th><th>conf</th>"
                f"<th>{_esc(primary_field)}</th>"
                f"<th>{_esc(secondary_field)}</th>"
                "<th>evidence (source · quote)</th>"
                "</tr></thead><tbody>"
            )
            for r in rules:
                if not isinstance(r, dict):
                    continue
                er_cells = []
                for er in (r.get("evidence_refs") or []):
                    if isinstance(er, dict):
                        er_cells.append(
                            f"<code>{_esc(er.get('source_ref', ''))}"
                            f"</code>: <em>{_esc(er.get('quote', ''))}</em>"
                        )
                er_html = "<br>".join(er_cells) or (
                    "<em>(no evidence_refs)</em>"
                )
                parts.append(
                    "<tr>"
                    f"<td><code>{_esc(r.get('rule_id', ''))}</code></td>"
                    f"<td>{_esc(r.get('confidence_band', ''))}</td>"
                    f"<td>{_esc(r.get(primary_field, ''))}</td>"
                    f"<td>{_esc(r.get(secondary_field, ''))}</td>"
                    f"<td>{er_html}</td>"
                    "</tr>"
                )
            parts.append("</tbody></table>")

        _render_rule_section(
            "must_stay_consistent",
            "§1.2 must_stay_consistent (keep across every shot)",
            "statement", "why_consistent",
        )
        _render_rule_section(
            "must_not_contradict",
            "§1.3 must_not_contradict (never introduce)",
            "do_not_introduce", "why",
        )
        # allowed_creative_freedom: include basis column.
        acf = evidence_pack.get("allowed_creative_freedom") or []
        if acf:
            parts.append(
                "<h3>§1.4 allowed_creative_freedom "
                f"(art-direction zones, count={len(acf)})</h3>"
            )
            parts.append(
                "<table><thead><tr>"
                "<th>rule_id</th><th>conf</th><th>basis</th>"
                "<th>free_to_choose</th><th>guidance</th>"
                "</tr></thead><tbody>"
            )
            for r in acf:
                if not isinstance(r, dict):
                    continue
                parts.append(
                    "<tr>"
                    f"<td><code>{_esc(r.get('rule_id', ''))}</code></td>"
                    f"<td>{_esc(r.get('confidence_band', ''))}</td>"
                    f"<td><code>{_esc(r.get('basis', ''))}</code></td>"
                    f"<td>{_esc(r.get('free_to_choose', ''))}</td>"
                    f"<td>{_esc(r.get('guidance', ''))}</td>"
                    "</tr>"
                )
            parts.append("</tbody></table>")
        # state_change_rules: list columns.
        scr = evidence_pack.get("state_change_rules") or []
        if scr:
            parts.append(
                "<h3>§1.5 state_change_rules "
                f"(per-event variation, count={len(scr)})</h3>"
            )
            parts.append(
                "<table><thead><tr>"
                "<th>rule_id</th><th>conf</th><th>change_kind</th>"
                "<th>when_applies</th><th>what_changes</th>"
                "<th>what_stays</th>"
                "</tr></thead><tbody>"
            )
            for r in scr:
                if not isinstance(r, dict):
                    continue
                parts.append(
                    "<tr>"
                    f"<td><code>{_esc(r.get('rule_id', ''))}</code></td>"
                    f"<td>{_esc(r.get('confidence_band', ''))}</td>"
                    f"<td>{_esc(r.get('change_kind', ''))}</td>"
                    f"<td>{_esc(r.get('when_applies', ''))}</td>"
                    f"<td>{_esc('; '.join(r.get('what_changes') or []))}</td>"
                    f"<td>{_esc('; '.join(r.get('what_stays') or []))}</td>"
                    "</tr>"
                )
            parts.append("</tbody></table>")
        # shot_conflict_checks.
        scc = evidence_pack.get("shot_conflict_checks") or []
        if scc:
            parts.append(
                "<h3>§1.6 shot_conflict_checks "
                f"(per-shot constraints, count={len(scc)})</h3>"
            )
            parts.append(
                "<p><em>Each row's evidence must include at least one "
                "quote whose source_ref equals the shot_ref "
                "(per-shot grounding). Episode-level quotes alone do "
                "not ground a per-shot conflict.</em></p>"
            )
            parts.append(
                "<table><thead><tr>"
                "<th>shot_ref</th><th>conf</th>"
                "<th>must_support</th><th>avoid_contradiction</th>"
                "<th>evidence (source · quote)</th>"
                "</tr></thead><tbody>"
            )
            for r in scc:
                if not isinstance(r, dict):
                    continue
                sref = r.get("shot_ref", "")
                er_cells = []
                for er in (r.get("evidence_refs") or []):
                    if not isinstance(er, dict):
                        continue
                    matches = er.get("source_ref") == sref
                    badge = (
                        " <strong style='color:#070'>[shot]</strong>"
                        if matches else ""
                    )
                    er_cells.append(
                        f"<code>{_esc(er.get('source_ref', ''))}"
                        f"</code>{badge}: "
                        f"<em>{_esc(er.get('quote', ''))}</em>"
                    )
                er_html = "<br>".join(er_cells) or (
                    "<em>(no evidence_refs)</em>"
                )
                parts.append(
                    "<tr>"
                    f"<td><code>{_esc(sref)}</code></td>"
                    f"<td>{_esc(r.get('confidence_band', ''))}</td>"
                    f"<td>{_esc('; '.join(r.get('must_support') or []))}</td>"
                    f"<td>{_esc('; '.join(r.get('avoid_contradiction') or []))}</td>"
                    f"<td>{er_html}</td>"
                    "</tr>"
                )
            parts.append("</tbody></table>")
        # §1.7 unknowns + coverage + rejected.
        unknowns = [
            u for u in (
                evidence_pack.get("unknowns_left_to_art_direction") or []
            )
            if isinstance(u, dict)
        ]
        if unknowns:
            parts.append("<h3>§1.7 unknowns_left_to_art_direction</h3><ul>")
            for u in unknowns:
                parts.append(
                    f"<li><strong>{_esc(u.get('area', ''))}</strong>: "
                    f"{_esc(u.get('note', ''))}</li>"
                )
            parts.append("</ul>")
        coverage = [
            c for c in (evidence_pack.get("coverage_notes") or [])
            if isinstance(c, dict)
        ]
        if coverage:
            parts.append("<h3>§1.8 coverage_notes</h3><ul>")
            for cn in coverage:
                parts.append(
                    f"<li><strong>{_esc(cn.get('note', ''))}</strong> — "
                    f"{_esc(cn.get('consequence_for_background', ''))} "
                    f"<em>({_esc(cn.get('confidence_band', ''))})</em></li>"
                )
            parts.append("</ul>")
        rejected = [
            r for r in (
                evidence_pack.get("rejected_or_irrelevant_summary") or []
            )
            if isinstance(r, dict)
        ]
        if rejected:
            parts.append(
                "<h3>§1.9 rejected_or_irrelevant_summary "
                "(caution reference only)</h3><ul>"
            )
            for rn in rejected:
                parts.append(
                    f"<li><strong>{_esc(rn.get('short_note', ''))}</strong>: "
                    f"{_esc(rn.get('why_not_pass_down', ''))}</li>"
                )
            parts.append("</ul>")
    parts.append(
        "<details><summary>Evidence Filter validation</summary>"
    )
    if not evidence_filter_report.get("executed", False):
        parts.append("<p><span class='badge'>planned only</span> "
                     "(generate off)</p><ul>")
        for c in evidence_filter_report.get("planned_checks", []):
            parts.append(f"<li>{_esc(c)}</li>")
        parts.append("</ul>")
    else:
        parts.append(
            f"<p>passed = <strong>"
            f"{evidence_filter_report.get('passed')}</strong></p>"
        )
        summary = evidence_filter_report.get("summary") or {}
        parts.append("<ul>")
        for k, v in summary.items():
            parts.append(
                f"<li><code>{_esc(k)}</code>: {_esc(v)}</li>"
            )
        parts.append("</ul>")
        if evidence_filter_report.get("failed_checks"):
            parts.append("<h4>failed_checks</h4><ul>")
            for f in evidence_filter_report["failed_checks"]:
                parts.append(f"<li class='bad'>{_esc(f)}</li>")
            parts.append("</ul>")
    parts.append("</details>")
    parts.append(
        "<details><summary>§2. Legacy Stage A / Stage B schemas "
        "(W2-simple: not active downstream — kept for compatibility)"
        "</summary>"
    )
    parts.append("<h3>Stage A — BackgroundWorldBrief schema</h3>")
    parts.append("<p>top-level: " + ", ".join(
        f"<code>{_esc(k)}</code>" for k in world_schema.get("required", [])
    ) + "</p>")
    parts.append("<p>world_background_brief subfields: " + ", ".join(
        f"<code>{_esc(k)}</code>"
        for k in world_schema.get("world_background_brief", {}).get("required", [])
    ) + "</p>")
    parts.append("<h3>Stage B — MinimalSpatialBrief schema</h3>")
    parts.append("<p>top-level: " + ", ".join(
        f"<code>{_esc(k)}</code>" for k in minimal_schema.get("required", [])
    ) + "</p>")
    parts.append("<p>visual_anchor.base_plate_role enum: " + ", ".join(
        f"<code>{_esc(r)}</code>" for r in BASE_PLATE_ROLES
    ) + " (identity_anchor means base plate should be informed by this "
                 "anchor, not that the anchor is rendered as an inventory item)</p>")
    parts.append("</details>")
    parts.append("<h2>§4. Prompt preview</h2>")
    parts.append("<h3>evidence_filter_system.txt (first 600 chars)</h3>")
    parts.append(
        f"<pre>{_esc(prompt_evidence_filter_system[:600])}</pre>"
    )
    parts.append("<h3>world_system.txt (first 600 chars)</h3>")
    parts.append(f"<pre>{_esc(prompt_world_system[:600])}</pre>")
    parts.append("<h3>spatial_system.txt (first 600 chars)</h3>")
    parts.append(f"<pre>{_esc(prompt_spatial_system[:600])}</pre>")
    parts.append("<h2>§5. Validation</h2>")
    if not validation_report.get("executed", False):
        parts.append("<p><span class='badge'>planned only</span> "
                     "checker not executed (--generate off).</p><ul>")
        for c in validation_report.get("planned_checks", []):
            parts.append(f"<li>{_esc(c)}</li>")
        parts.append("</ul>")
        if validation_report.get("note"):
            parts.append(f"<p><em>{_esc(validation_report['note'])}</em></p>")
    else:
        parts.append(
            f"<p>passed = <strong>{validation_report.get('passed')}</strong></p>"
        )
        summary = validation_report.get("summary") or {}
        parts.append("<ul>")
        for k, v in summary.items():
            parts.append(f"<li><code>{_esc(k)}</code>: {_esc(v)}</li>")
        parts.append("</ul>")
        if validation_report.get("failed_checks"):
            parts.append("<h3>failed_checks</h3><ul>")
            for f in validation_report["failed_checks"]:
                parts.append(f"<li class='bad'>{_esc(f)}</li>")
            parts.append("</ul>")
    # ── §99. Raw SourceBundle preview (collapsed at bottom by design) ──
    parts.append("<h2>§99. Raw SourceBundle (read only by Evidence Filter)</h2>")
    parts.append(
        f"<p>total sources = <strong>{len(bundle.sources)}</strong>"
        f" / char_budget_estimate = <strong>{bundle.char_budget_estimate}</strong>"
        f". <em>This is NOT the downstream input. Only the Evidence "
        f"Filter LLM reads this; the world / spatial stages consume "
        f"only the refined pack above (§1).</em></p>"
    )
    parts.append(
        "<details><summary>Show raw SourceBundle table</summary>"
    )
    parts.append("<table><thead><tr><th>source_ref</th><th>kind</th>"
                 "<th>sha256</th><th>char_count</th><th>preview</th></tr>"
                 "</thead><tbody>")
    for src in bundle.sources:
        preview = src.text[:200].replace("\n", " ")
        parts.append(
            f"<tr><td><code>{_esc(src.source_ref)}</code></td>"
            f"<td><code>{_esc(src.kind)}</code></td>"
            f"<td><code>{_esc(src.sha256[:16])}…</code></td>"
            f"<td>{src.char_count}</td>"
            f"<td>{_esc(preview)}</td></tr>"
        )
    parts.append("</tbody></table></details>")
    parts.append("</body></html>")
    return "".join(parts)


# ============================================================================
# Output writers (13 files)
# ============================================================================
def _bundle_to_dict(bundle: SourceBundle) -> dict:
    return {
        "bundle_id": bundle.bundle_id,
        "fixture_id": bundle.fixture_id,
        "project_id": bundle.project_id,
        "episode_id": bundle.episode_id,
        "location_short_id": bundle.location_short_id,
        "char_budget_estimate": bundle.char_budget_estimate,
        "sources": [
            {
                "source_ref": s.source_ref,
                "kind": s.kind,
                "text": s.text,
                "sha256": s.sha256,
                "char_count": s.char_count,
                **({k: v for k, v in s.extras.items()} if s.extras else {}),
            }
            for s in bundle.sources
        ],
        "missing_inputs": list(bundle.missing_inputs),
    }


def _render_source_bundle_md(bundle: SourceBundle) -> str:
    lines: list[str] = [
        f"# SourceBundle — {bundle.bundle_id}", "",
        f"- fixture_id: `{bundle.fixture_id}`",
        f"- project_id: `{bundle.project_id}`",
        f"- episode_id: `{bundle.episode_id}`",
        f"- location_short_id: `{bundle.location_short_id}`",
        f"- total sources: {len(bundle.sources)}",
        f"- char_budget_estimate: {bundle.char_budget_estimate}",
        "",
    ]
    for src in bundle.sources:
        lines.append(f"## `{src.source_ref}` ({src.kind})")
        lines.append("")
        lines.append(f"- sha256: `{src.sha256}`")
        lines.append(f"- char_count: {src.char_count}")
        if src.extras:
            lines.append(f"- extras: `{json.dumps(src.extras, ensure_ascii=False)}`")
        preview = src.text[:300].replace("\n", " ")
        lines.append("")
        lines.append(f"> {preview}")
        lines.append("")
    return "\n".join(lines)


def write_outputs(*, run_dir: Path, bundle: SourceBundle,
                   world_schema: dict, minimal_schema: dict,
                   evidence_pack_schema: dict, evidence_pack: dict,
                   evidence_filter_report: dict,
                   prompt_evidence_filter_system: str,
                   prompt_evidence_filter_user: str,
                   prompt_world_system: str, prompt_world_user: str,
                   prompt_spatial_system: str, prompt_spatial_user: str,
                   world_brief: dict, minimal_brief: dict,
                   validation_report: dict, html: str,
                   run_meta: dict,
                   quarantined_pack: Optional[dict] = None) -> None:
    run_dir.mkdir(parents=True, exist_ok=True)
    prompt_dir = run_dir / "prompt"
    prompt_dir.mkdir(parents=True, exist_ok=True)
    (run_dir / "source_bundle.json").write_text(
        json.dumps(_bundle_to_dict(bundle), ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (run_dir / "source_bundle.md").write_text(
        _render_source_bundle_md(bundle), encoding="utf-8",
    )
    (run_dir / "background_evidence_filter_schema.json").write_text(
        json.dumps(evidence_pack_schema, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (run_dir / "world_brief_schema.json").write_text(
        json.dumps(world_schema, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (run_dir / "minimal_spatial_brief_schema.json").write_text(
        json.dumps(minimal_schema, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (prompt_dir / "evidence_filter_system.txt").write_text(
        prompt_evidence_filter_system, encoding="utf-8",
    )
    (prompt_dir / "evidence_filter_user.txt").write_text(
        prompt_evidence_filter_user, encoding="utf-8",
    )
    (prompt_dir / "world_system.txt").write_text(
        prompt_world_system, encoding="utf-8")
    (prompt_dir / "world_user.txt").write_text(
        prompt_world_user, encoding="utf-8")
    (prompt_dir / "spatial_system.txt").write_text(
        prompt_spatial_system, encoding="utf-8")
    (prompt_dir / "spatial_user.txt").write_text(
        prompt_spatial_user, encoding="utf-8")
    (run_dir / "background_evidence_pack.json").write_text(
        json.dumps(evidence_pack, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    if quarantined_pack is not None:
        (run_dir / "background_evidence_pack_quarantined.json").write_text(
            json.dumps(quarantined_pack, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
    (run_dir / "world_brief.json").write_text(
        json.dumps(world_brief, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (run_dir / "minimal_spatial_brief.json").write_text(
        json.dumps(minimal_brief, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (run_dir / "filter_validation_report.json").write_text(
        json.dumps(evidence_filter_report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (run_dir / "validation_report.json").write_text(
        json.dumps(validation_report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    (run_dir / "index.html").write_text(html, encoding="utf-8")
    (run_dir / "run_meta.json").write_text(
        json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8",
    )


# ============================================================================
# Diagnostic artifact loader (default off, deprecated marker)
# ============================================================================
def _resolve_latest_run(parent: Path) -> Optional[Path]:
    if not parent.exists() or not parent.is_dir():
        return None
    candidates = []
    for p in parent.iterdir():
        if not p.is_dir():
            continue
        name = p.name
        if len(name) < 13 or name[8] != "_" or name[13] != "_":
            continue
        if not (name[:8].isdigit() and name[9:13].isdigit()):
            continue
        candidates.append(p)
    candidates.sort()
    return candidates[-1] if candidates else None


def load_diagnostic_artifacts(repo_root: Path) -> list[tuple[str, Any]]:
    parents = [
        Path("scripts_output/background_place_grouping_experiment"),
        Path("scripts_output/background_spatial_decision_experiment"),
        Path("scripts_output/background_topology_planner_experiment"),
    ]
    files = [
        "source_evidence_pack.json", "source_coverage_summary.json",
        "place_groups.json", "set_groups.json", "shot_bindings.json",
    ]
    out: list[tuple[str, Any]] = []
    for parent_rel in parents:
        parent_abs = repo_root / parent_rel
        latest = _resolve_latest_run(parent_abs)
        if latest is None:
            continue
        for fname in files:
            fp = latest / fname
            if not fp.exists():
                continue
            try:
                obj = json.loads(fp.read_text(encoding="utf-8"))
            except (OSError, ValueError):
                continue
            out.append((str(fp.relative_to(repo_root)), obj))
    return out


# ============================================================================
# Run meta / id
# ============================================================================
def _now_iso() -> str:
    return datetime.now(timezone(timedelta(hours=9))).isoformat(timespec="seconds")


def _short_uuid() -> str:
    return uuid.uuid4().hex[:6]


def _new_run_id() -> str:
    return (
        datetime.now(timezone(timedelta(hours=9))).strftime("%Y%m%d_%H%M")
        + "_" + _short_uuid()
    )


def build_run_meta(*, run_id: str, fixture_id: str, args_dict: dict,
                    outputs: list[str], model: str,
                    char_budget_estimate: int, bundle_sha256: str,
                    run_status: str = "dry_run",
                    exit_code: int = 0,
                    filter_validation_passed: Optional[bool] = None,
                    generate_status: Optional[str] = None,
                    filter_failed_checks: Optional[list[str]] = None,
                    initial_failed_checks: Optional[list[str]] = None,
                    quarantine_output: Optional[str] = None,
                    correction_attempted: bool = False,
                    correction_status: str = "not_attempted",
                    correction_model: Optional[str] = None,
                    correction_failed_checks: Optional[list[str]] = None,
                    evidence_filter_attempts: int = 0) -> dict:
    """W2a-c: run_meta now carries the actual run status so that
    downstream code and the HTML cockpit never mistake a fail-closed
    generate run for a clean dry-run.

    `filter_validation_passed` is intentionally tri-state: None for
    dry-run (checker not executed), True/False for generate runs.
    """
    return {
        "run_id": run_id,
        "plan_version": PLAN_VERSION,
        "generated_at": _now_iso(),
        "fixture_id": fixture_id,
        "model": model,
        "stages": [
            "background_continuity_brief",
            "_legacy_world_brief_placeholder",
            "_legacy_minimal_spatial_brief_placeholder",
        ],
        "args": args_dict,
        "outputs": list(outputs),
        "char_budget_estimate": char_budget_estimate,
        "sha256": bundle_sha256,
        "run_status": run_status,
        "exit_code": int(exit_code),
        "filter_validation_passed": filter_validation_passed,
        "generate_status": generate_status,
        "filter_failed_checks": list(filter_failed_checks or []),
        "initial_failed_checks": list(initial_failed_checks or []),
        "quarantine_output": quarantine_output,
        "correction_attempted": bool(correction_attempted),
        "correction_status": correction_status,
        "correction_model": correction_model,
        "correction_failed_checks": list(correction_failed_checks or []),
        "evidence_filter_attempts": int(evidence_filter_attempts),
    }


# ============================================================================
# CLI + main
# ============================================================================
def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:
    ap = argparse.ArgumentParser(
        description=(
            "background semantic extractor experiment (W2a-d). 3-stage "
            "LLM: Evidence Filter (gateway over the SourceBundle) → "
            "world brief → minimal spatial brief. Downstream stages "
            "consume only the filtered BackgroundEvidencePack. "
            "--generate default off; fail-closed on validation failure "
            "with a single LLM correction retry for quote_not_in_source "
            "failures (deterministic code never paraphrases quotes). "
            "Invalid pack quarantined, exit code 1. No lexicon / regex "
            "/ boundary / particle logic anywhere."
        ),
    )
    ap.add_argument("--output-root", type=Path,
                    default=_REPO_ROOT / DEFAULT_OUTPUT_DIR)
    ap.add_argument("--run-id", type=str, default=None)
    ap.add_argument("--project-id", type=str, default=None)
    ap.add_argument("--episode-id", type=str, default=None)
    ap.add_argument("--location-short-id", type=str, default=None)
    ap.add_argument("--include-diagnostic", action="store_true", default=False)
    ap.add_argument("--generate", action="store_true", default=False)
    ap.add_argument("--dry-run", action="store_true", default=False,
                    help="Explicit dry-run; forces --generate off.")
    ap.add_argument("--model", type=str, default=None)
    ap.add_argument("--no-serve", action="store_true", default=False)
    ap.add_argument("--serve-port", type=int, default=None)
    return ap.parse_args(argv)


def _resolve_effective_args(args: argparse.Namespace,
                             spec: SampleFixtureSpec) -> dict[str, Any]:
    generate = bool(args.generate) and not bool(args.dry_run)
    model = args.model or spec.default_model
    return {
        "output_root": str(args.output_root),
        "run_id": args.run_id,
        "project_id": args.project_id,
        "episode_id": args.episode_id,
        "location_short_id": args.location_short_id,
        "include_diagnostic": bool(args.include_diagnostic),
        "generate": generate,
        "dry_run": bool(args.dry_run),
        "model": model,
        "no_serve": bool(args.no_serve),
        "serve_port": args.serve_port,
    }


def main(argv: Optional[list[str]] = None) -> int:
    args = parse_args(argv)
    spec = build_sample_fixture_l05_spec()
    if args.project_id and args.project_id != spec.project_id:
        spec = SampleFixtureSpec(
            **{**spec.__dict__, "project_id": args.project_id,
               "fixture_id": f"{spec.fixture_id}_override"},
        )
    if args.episode_id and args.episode_id != spec.episode_id:
        spec = SampleFixtureSpec(
            **{**spec.__dict__, "episode_id": args.episode_id,
               "fixture_id": f"{spec.fixture_id}_override"},
        )
    if args.location_short_id and args.location_short_id != spec.location_short_id:
        spec = SampleFixtureSpec(
            **{**spec.__dict__,
               "location_short_id": args.location_short_id,
               "fixture_id": f"{spec.fixture_id}_override"},
        )
    effective = _resolve_effective_args(args, spec)
    run_id = args.run_id or _new_run_id()
    out_dir = args.output_root / run_id

    # ---- DB read-only ------------------------------------------------------
    from app.core.database import SessionLocal
    missing_inputs: list[dict] = []
    with SessionLocal() as session:
        plan_text, plan_meta = load_planning_doc(session, spec.project_id)
        ep_text, ep_meta = load_episode_fulltext(session, spec.episode_id)
        shots = load_selected_shots_for_location(
            session, spec.project_id, spec.episode_id, spec.location_short_id,
        )
        entities = load_entity_catalog(session, spec.project_id)
        locations = load_location_catalog(session, spec.project_id)
    if plan_meta.get("missing"):
        missing_inputs.append({"key": "planning_doc", "id": plan_meta.get("id")})
    if ep_meta.get("missing"):
        missing_inputs.append({"key": "episode_fulltext", "id": ep_meta.get("id")})

    diagnostic_artifacts: list[tuple[str, Any]] = []
    if effective["include_diagnostic"]:
        diagnostic_artifacts = load_diagnostic_artifacts(_REPO_ROOT)

    bundle = build_source_bundle(
        spec=spec, run_id=run_id,
        planning_text=plan_text, episode_text=ep_text,
        selected_shots=shots, entity_catalog=entities,
        location_catalog=locations,
        diagnostic_artifacts=diagnostic_artifacts,
        missing_inputs=missing_inputs,
    )
    world_schema = build_world_brief_schema(bundle)
    minimal_schema = build_minimal_spatial_brief_schema(bundle)
    evidence_pack_schema = build_background_evidence_pack_schema(bundle)
    prompt_evidence_filter_system = build_prompt_evidence_filter_system(spec)
    prompt_evidence_filter_user = build_prompt_evidence_filter_user(
        spec=spec, bundle=bundle, schema=evidence_pack_schema,
    )
    prompt_world_system = build_prompt_world_system(spec)
    prompt_spatial_system = build_prompt_spatial_system(spec)

    # Evidence Filter stage (W2a + W2a-b). The route_evidence_filter_
    # result() helper separates "LLM call succeeded" from "EvidencePack is
    # valid"; an invalid pack is quarantined and a placeholder takes its
    # place for downstream prompts.
    evidence_pack: dict
    evidence_filter_report: dict
    quarantined_pack: Optional[dict] = None
    exit_code = 0
    run_status = "dry_run"
    if effective["generate"]:
        check_char_budget_fail_closed(bundle=bundle, spec=spec)
        filter_result = run_evidence_filter_generate(
            bundle=bundle, schema=evidence_pack_schema,
            model=effective["model"] or EVIDENCE_PACK_DEFAULT_GENERATE_MODEL,
            llm_call=_default_evidence_filter_llm_call,
            max_retries=1,
        )
        routed = route_evidence_filter_result(
            filter_result=filter_result, bundle=bundle,
            schema=evidence_pack_schema,
            correction_call=_default_evidence_filter_correction_llm_call,
            model=(effective["model"]
                   or EVIDENCE_PACK_DEFAULT_GENERATE_MODEL),
        )
        evidence_pack = routed["downstream_pack"]
        evidence_filter_report = routed["evidence_filter_report"]
        quarantined_pack = routed["quarantined_pack"]
        run_status = routed["run_status"]
        exit_code = routed["exit_code"]
    else:
        evidence_pack = placeholder_evidence_pack(bundle)
        evidence_filter_report = planned_evidence_filter_report()

    prompt_world_user = build_prompt_world_user(
        spec=spec, evidence_pack=evidence_pack, schema=world_schema,
    )

    # Stage B prompt builder receives ONLY world_brief_ref +
    # world_hints_for_background (plan §6-E-22 W0g BLOCKING) plus the
    # filtered evidence_pack (W2a).
    placeholder_world_brief_ref = "world_background_brief"
    placeholder_world_hints: list[str] = []
    prompt_spatial_user = build_prompt_spatial_user(
        spec=spec, evidence_pack=evidence_pack, schema=minimal_schema,
        world_brief_ref=placeholder_world_brief_ref,
        world_hints_for_background=placeholder_world_hints,
    )

    world_brief: dict
    minimal_brief: dict
    if effective["generate"]:
        # W2 will replace these placeholders with actual world/spatial LLM
        # responses. Today --generate exercises only the Evidence Filter.
        world_brief = {
            "placeholder": True, "stage": "world",
            "reason": "world brief stage not yet wired into --generate",
        }
        minimal_brief = {
            "placeholder": True, "stage": "spatial",
            "reason": "spatial brief stage not yet wired into --generate",
        }
        stage_b_input = {
            "world_brief_ref": placeholder_world_brief_ref,
            "world_hints_for_background": placeholder_world_hints,
        }
        validation = run_validation(
            world_brief=world_brief, minimal_brief=minimal_brief,
            stage_b_input=stage_b_input, bundle=bundle, spec=spec,
            world_schema=world_schema, minimal_schema=minimal_schema,
        )
    else:
        world_brief = {
            "placeholder": True, "stage": "world",
            "reason": "dry-run only; --generate default off",
            "schema_version": STAGE_A_SCHEMA_VERSION,
        }
        minimal_brief = {
            "placeholder": True, "stage": "spatial",
            "reason": "dry-run only; --generate default off",
            "schema_version": STAGE_B_SCHEMA_VERSION,
        }
        validation = planned_checker_report()

    bundle_sha = _sha256_hex(json.dumps(
        _bundle_to_dict(bundle), ensure_ascii=False, sort_keys=True,
    ))
    outputs_list = [
        "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",
    ]
    if quarantined_pack is not None:
        outputs_list.append("background_evidence_pack_quarantined.json")
    quarantine_output_name = (
        "background_evidence_pack_quarantined.json"
        if quarantined_pack is not None else None
    )
    filter_validation_passed: Optional[bool]
    if run_status == "dry_run":
        filter_validation_passed = None
    else:
        filter_validation_passed = bool(
            evidence_filter_report.get("passed")
        )
    run_meta = build_run_meta(
        run_id=run_id, fixture_id=spec.fixture_id, args_dict=effective,
        outputs=outputs_list,
        model=effective["model"],
        char_budget_estimate=bundle.char_budget_estimate,
        bundle_sha256=bundle_sha,
        run_status=run_status,
        exit_code=exit_code,
        filter_validation_passed=filter_validation_passed,
        generate_status=evidence_filter_report.get("generate_status"),
        filter_failed_checks=(
            evidence_filter_report.get("failed_checks") or []
        ),
        initial_failed_checks=(
            evidence_filter_report.get("initial_failed_checks") or []
        ),
        quarantine_output=quarantine_output_name,
        correction_attempted=bool(
            evidence_filter_report.get("correction_attempted")
        ),
        correction_status=(
            evidence_filter_report.get("correction_status")
            or "not_attempted"
        ),
        correction_model=evidence_filter_report.get("correction_model"),
        correction_failed_checks=evidence_filter_report.get(
            "correction_failed_checks") or [],
        evidence_filter_attempts=int(
            evidence_filter_report.get("evidence_filter_attempts") or 0
        ),
    )
    html = render_html(
        bundle=bundle, world_schema=world_schema,
        minimal_schema=minimal_schema,
        evidence_pack_schema=evidence_pack_schema,
        evidence_pack=evidence_pack,
        evidence_filter_report=evidence_filter_report,
        prompt_evidence_filter_system=prompt_evidence_filter_system,
        prompt_evidence_filter_user=prompt_evidence_filter_user,
        prompt_world_system=prompt_world_system,
        prompt_world_user=prompt_world_user,
        prompt_spatial_system=prompt_spatial_system,
        prompt_spatial_user=prompt_spatial_user,
        validation_report=validation, run_meta=run_meta,
        include_diagnostic=effective["include_diagnostic"],
    )
    write_outputs(
        run_dir=out_dir, bundle=bundle,
        world_schema=world_schema, minimal_schema=minimal_schema,
        evidence_pack_schema=evidence_pack_schema,
        evidence_pack=evidence_pack,
        evidence_filter_report=evidence_filter_report,
        prompt_evidence_filter_system=prompt_evidence_filter_system,
        prompt_evidence_filter_user=prompt_evidence_filter_user,
        prompt_world_system=prompt_world_system,
        prompt_world_user=prompt_world_user,
        prompt_spatial_system=prompt_spatial_system,
        prompt_spatial_user=prompt_spatial_user,
        world_brief=world_brief, minimal_brief=minimal_brief,
        validation_report=validation, html=html, run_meta=run_meta,
        quarantined_pack=quarantined_pack,
    )
    print(f"[bsx] run_id={run_id}")
    print(f"[bsx] out_dir={out_dir}")
    print(f"[bsx] sources={len(bundle.sources)} "
          f"char_budget={bundle.char_budget_estimate} "
          f"generate={effective['generate']} "
          f"include_diagnostic={effective['include_diagnostic']}")
    print(f"[bsx] run_status={run_status} exit_code={exit_code}")
    if quarantined_pack is not None:
        print(
            "[bsx] quarantined invalid evidence pack saved to "
            "background_evidence_pack_quarantined.json — downstream "
            "prompts used placeholder"
        )
    return exit_code


if __name__ == "__main__":
    sys.exit(main())