"""experiment_floor_plan_downstream_payload_preview_slice — W17E.

Consume the W17D coarse-layout synthesis JSON (and the W17C → W17B3 →
W17A → W15e chain) and emit a downstream-payload preview + guard-
consumption dry-run. This stage defines the deterministic contract by
which a downstream consumer (eventually W13/W14 background render /
prompt assembly) would join the W15e candidate truth with the W17D
synthesis observation and decide which markers must BLOCK / IMPORTANT /
MINOR / OK.

No LLM call, no VLM call, no image API call. No DB write. No
production manifest mutation. No commit / push. Pure JSON-to-JSON
consumption.

CLI:
  --derive-payload-from <W17D_run_dir>          (required)
  --target-fp-ids fp_l05_01                     (default, restricted)
  --output-root <path>
  --diag-print-imports
"""
from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple

_REPO_ROOT = Path(__file__).resolve().parents[2]
_SCRIPTS_DIR = _REPO_ROOT / "backend" / "scripts"
if str(_SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS_DIR))

from experiment_background_pipeline_slice import (  # type: ignore
    KST,
    PLAN_VERSION,
    _check_image_imports_present,
    _check_production_diff_empty,
    _maybe_print_imports,
)
from experiment_floor_plan_grid_layout_slice import (  # type: ignore
    W16_ALLOWED_TARGET_FP_IDS,
    _resolve_targets,
)

W17E_STAGE = "w17e_floor_plan_downstream_payload_preview_slice"

# guard_level enum — strict whitelist, deterministic policy preview only.
# Not a semantic proof; W15e candidate stays the truth surface.
W17E_GUARD_LEVEL_ENUM = ("BLOCKING", "IMPORTANT", "MINOR", "OK")
W17E_SOURCE_TRUTH = "w15e_candidate"
W17E_OBSERVATION_SOURCE = "w17d_synthesis"

_DEFAULT_OUTPUT_ROOT = (
    _REPO_ROOT / "scripts_output" / "floor_plan_downstream_payload_preview_slice_experiment"
)


def _run_id() -> str:
    import secrets

    return datetime.now(KST).strftime("%Y%m%d_%H%M") + "_" + secrets.token_hex(3)


def _parse_args(argv):
    p = argparse.ArgumentParser(
        description=(
            "W17E downstream payload preview — deterministic only. "
            "No LLM/VLM/image API call."
        )
    )
    p.add_argument(
        "--derive-payload-from", required=True,
        help="Path to a prior W17D success run dir.",
    )
    p.add_argument(
        "--target-fp-ids", default="fp_l05_01",
        help="Comma-separated fp_id subset. Wave 1 only allows fp_l05_01.",
    )
    p.add_argument("--output-root", default=str(_DEFAULT_OUTPUT_ROOT))
    p.add_argument("--diag-print-imports", action="store_true")
    return p.parse_args(argv)


# ─────────────────────────────────────────────────────────────────────────────
# Chain resolver (W17D → W17C → W17B3 → W17A → W15e) for legend priority join
# ─────────────────────────────────────────────────────────────────────────────

def _resolve_chain_run_dir(
    *, current_meta: Dict[str, Any], arg_key: str,
    stage_parent_subdir: str, required_file: str,
) -> Optional[Path]:
    args_dict = (current_meta or {}).get("args") or {}
    if isinstance(args_dict, dict):
        raw = args_dict.get(arg_key)
        if isinstance(raw, str) and raw:
            p = Path(raw)
            if not p.is_absolute():
                p = _REPO_ROOT / raw
            if (p / required_file).exists():
                return p
    derived_from = (current_meta or {}).get("derived_from") or ""
    if isinstance(derived_from, str) and derived_from:
        parent = _REPO_ROOT / "scripts_output" / stage_parent_subdir
        if (parent / derived_from / required_file).exists():
            return parent / derived_from
    return None


def _load_w17a_legend_priority_index(
    w17d_run_dir: Path,
) -> Dict[str, Dict[int, str]]:
    """Walk W17D → W17C → W17B3 → W17A and return
    `{fp_id: {marker_number: priority_str}}`. Empty dict on any
    missing link — caller must tolerate (downstream invariants
    surface that via `inputs_present`)."""
    out: Dict[str, Dict[int, str]] = {}
    try:
        w17d_meta = json.loads((w17d_run_dir / "run_meta.json").read_text())
    except Exception:  # noqa: BLE001
        return out
    w17c_dir = _resolve_chain_run_dir(
        current_meta=w17d_meta,
        arg_key="derive_synthesis_from",
        stage_parent_subdir="floor_plan_vlm_readback_slice_experiment",
        required_file="run_meta.json",
    )
    if w17c_dir is None:
        return out
    try:
        w17c_meta = json.loads((w17c_dir / "run_meta.json").read_text())
    except Exception:  # noqa: BLE001
        return out
    w17b3_dir = _resolve_chain_run_dir(
        current_meta=w17c_meta,
        arg_key="derive_readback_from",
        stage_parent_subdir="floor_plan_image_smoke_slice_experiment",
        required_file="run_meta.json",
    )
    if w17b3_dir is None:
        return out
    try:
        w17b3_meta = json.loads((w17b3_dir / "run_meta.json").read_text())
    except Exception:  # noqa: BLE001
        return out
    w17a_dir = _resolve_chain_run_dir(
        current_meta=w17b3_meta,
        arg_key="derive_image_smoke_from",
        stage_parent_subdir="floor_plan_image_prompt_slice_experiment",
        required_file="floor_plan_image_prompt.json",
    )
    if w17a_dir is None:
        return out
    try:
        prompt_artifact = json.loads(
            (w17a_dir / "floor_plan_image_prompt.json").read_text()
        )
    except Exception:  # noqa: BLE001
        return out
    by_fp = (prompt_artifact or {}).get("floor_plan_image_prompt_by_fp") or {}
    for fp_id, fp_entry in by_fp.items():
        if not isinstance(fp_entry, dict):
            continue
        priority_by_n: Dict[int, str] = {}
        for it in fp_entry.get("numbered_marker_legend") or []:
            if isinstance(it, dict) and isinstance(it.get("marker_number"), int):
                priority_by_n[int(it["marker_number"])] = it.get("priority") or ""
        out[fp_id] = priority_by_n
    return out


# ─────────────────────────────────────────────────────────────────────────────
# Guard level policy (deterministic preview, not semantic proof)
# ─────────────────────────────────────────────────────────────────────────────

def _classify_guard_level(
    *, marker: Dict[str, Any],
    priority: str,
    vlm_reported_conflict: bool,
) -> Tuple[str, str]:
    """Return (guard_level, guard_reason) per Codex W17E spec."""
    bucket = marker.get("bucket") or ""
    visible = bool(marker.get("visible"))
    confidence = (marker.get("confidence") or "").lower()
    heuristic_match = bool(marker.get("heuristic_unit_match"))
    region = (marker.get("approximate_region_or_cell") or "").strip()
    notes = (marker.get("notes") or "").lower()

    # BLOCKING precedence first.
    if bucket in ("missing", "unaccounted_in_readback"):
        return ("BLOCKING", f"bucket={bucket}; marker not present in readback")
    if not visible:
        return ("BLOCKING", "visible=false")
    if (priority or "").lower() == "must_show" and bucket != "matched":
        return (
            "BLOCKING",
            f"priority=must_show but bucket={bucket}",
        )
    if vlm_reported_conflict:
        return ("BLOCKING", "VLM-reported readback conflict")

    # IMPORTANT.
    if bucket == "matched" and confidence != "high":
        return ("IMPORTANT", f"matched but confidence={confidence or 'unknown'}")
    if not heuristic_match and not vlm_reported_conflict:
        return (
            "IMPORTANT",
            "observed_unit does not heuristically contain expected_unit; "
            "no VLM conflict reported",
        )
    if not region:
        return ("IMPORTANT", "approximate_region_or_cell empty or vague")

    # MINOR — matched + high + heuristic ok, but notes mention review words.
    uncertainty_tokens = ("uncertain", "verify", "review", "optional", "fuzzy")
    if any(tok in notes for tok in uncertainty_tokens):
        return (
            "MINOR",
            f"matched + high confidence; notes mention review/uncertainty",
        )

    # OK.
    return (
        "OK",
        "matched + high confidence + heuristic_unit_match + no VLM conflict",
    )


# ─────────────────────────────────────────────────────────────────────────────
# Payload + dry-run consumption builders
# ─────────────────────────────────────────────────────────────────────────────

def _build_payload_for_fp(
    *, fp_id: str,
    synthesis_entry: Dict[str, Any],
    mismatch_entry: Dict[str, Any],
    priority_by_marker: Dict[int, str],
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """Return (payload_entry, consumption_trace_entry) for one fp."""
    vlm_conflict_numbers: Set[int] = set()
    for c in mismatch_entry.get("vlm_reported_conflicts") or []:
        if isinstance(c, dict) and isinstance(c.get("marker_number"), int):
            vlm_conflict_numbers.add(int(c["marker_number"]))

    markers_out: List[Dict[str, Any]] = []
    level_counts: Dict[str, int] = {lv: 0 for lv in W17E_GUARD_LEVEL_ENUM}

    for marker in synthesis_entry.get("markers") or []:
        if not isinstance(marker, dict):
            continue
        n = marker.get("marker_number")
        if not isinstance(n, int):
            continue
        priority = priority_by_marker.get(n, "")
        vlm_conflict = n in vlm_conflict_numbers
        level, reason = _classify_guard_level(
            marker=marker, priority=priority,
            vlm_reported_conflict=vlm_conflict,
        )
        level_counts[level] = level_counts.get(level, 0) + 1
        markers_out.append({
            "marker_number": n,
            "label": marker.get("label") or "",
            "element_kind": marker.get("kind") or "",
            "visual_encoding": marker.get("visual_encoding") or "",
            "expected_unit": marker.get("expected_unit") or "",
            "expected_placement": marker.get("expected_placement") or "",
            "observed_unit_label_or_area": marker.get(
                "observed_unit_label_or_area"
            ) or "",
            "approximate_region_or_cell": marker.get(
                "approximate_region_or_cell"
            ) or "",
            "visible": bool(marker.get("visible")),
            "confidence": marker.get("confidence") or "",
            "bucket": marker.get("bucket") or "",
            "priority": priority,
            "vlm_reported_conflict": vlm_conflict,
            "guard_level": level,
            "guard_reason": reason,
            "source_truth": W17E_SOURCE_TRUTH,
            "observation_source": W17E_OBSERVATION_SOURCE,
        })

    payload_entry = {
        "fp_id": fp_id,
        "markers": markers_out,
        "unit_boundary_summary": synthesis_entry.get("unit_boundary_summary") or "",
        "guard_level_counts": dict(level_counts),
        "source_truth": W17E_SOURCE_TRUTH,
        "observation_source": W17E_OBSERVATION_SOURCE,
    }

    # Dry-run consumption trace — what a downstream consumer would do
    # with this payload deterministically.
    blocking_markers = [m for m in markers_out if m["guard_level"] == "BLOCKING"]
    important_markers = [m for m in markers_out if m["guard_level"] == "IMPORTANT"]
    consumption_trace = {
        "fp_id": fp_id,
        "BLOCKING_count": len(blocking_markers),
        "IMPORTANT_count": len(important_markers),
        "MINOR_count": sum(1 for m in markers_out if m["guard_level"] == "MINOR"),
        "OK_count": sum(1 for m in markers_out if m["guard_level"] == "OK"),
        "BLOCKING_marker_numbers": [m["marker_number"] for m in blocking_markers],
        "IMPORTANT_marker_numbers": [m["marker_number"] for m in important_markers],
        "consumer_decision_dry_run": (
            "downstream_REJECT_render_if_any_BLOCKING"
            if blocking_markers
            else "downstream_PROCEED_render_with_observation_diagnostics"
        ),
        "policy_note": (
            "guard_level is a deterministic policy preview, not a semantic "
            "proof. W15e candidate stays the truth surface; W17D synthesis "
            "is an observation layer."
        ),
    }
    return payload_entry, consumption_trace


# ─────────────────────────────────────────────────────────────────────────────
# Compatibility report
# ─────────────────────────────────────────────────────────────────────────────

def _build_w17e_compatibility_report(
    *, payload_by_fp: Dict[str, Dict[str, Any]],
    mismatch_by_fp: Dict[str, Dict[str, Any]],
    synthesis_by_fp: Dict[str, Dict[str, Any]],
    priority_by_fp: Dict[str, Dict[int, str]],
    production_diff_empty: bool,
    db_write_count: int,
    image_import_seen: bool,
    image_api_call_count: int,
    vlm_api_call_count: int,
    llm_api_call_count: int,
    missing_inputs: List[str],
    prev_run_id: str,
    target_fp_ids: Set[str],
) -> dict:
    inv: Dict[str, Dict[str, Any]] = {}

    inv["inputs_present"] = {
        "pass": (
            not missing_inputs
            and bool(payload_by_fp)
            and bool(synthesis_by_fp)
        ),
        "detail": {
            "missing_inputs": list(missing_inputs),
            "payload_fp_count": len(payload_by_fp),
            "synthesis_fp_count": len(synthesis_by_fp),
            "prev_run_id": prev_run_id,
        },
    }

    # payload_covers_full_synthesis_and_legend
    coverage_failures: List[dict] = []
    for fp_id, synth in synthesis_by_fp.items():
        synth_numbers = {
            int(m["marker_number"]) for m in (synth.get("markers") or [])
            if isinstance(m, dict) and isinstance(m.get("marker_number"), int)
        }
        legend_numbers = set(priority_by_fp.get(fp_id, {}).keys())
        payload = payload_by_fp.get(fp_id) or {}
        payload_numbers = {
            int(m["marker_number"]) for m in (payload.get("markers") or [])
            if isinstance(m, dict) and isinstance(m.get("marker_number"), int)
        }
        if payload_numbers != synth_numbers:
            coverage_failures.append({
                "fp_id": fp_id,
                "reason": "payload != synthesis",
                "payload_extras": sorted(payload_numbers - synth_numbers),
                "synthesis_extras": sorted(synth_numbers - payload_numbers),
            })
        if legend_numbers and payload_numbers != legend_numbers:
            coverage_failures.append({
                "fp_id": fp_id,
                "reason": "payload != legend",
                "missing_in_payload": sorted(legend_numbers - payload_numbers),
                "extras_in_payload": sorted(payload_numbers - legend_numbers),
            })
    inv["payload_covers_full_synthesis_and_legend"] = {
        "pass": not coverage_failures,
        "detail": {
            "failures": coverage_failures[:30],
            "failure_count": len(coverage_failures),
        },
    }

    # guard_level_enum_and_required_reason_present
    enum_failures: List[dict] = []
    for fp_id, payload in payload_by_fp.items():
        for m in payload.get("markers") or []:
            if m.get("guard_level") not in W17E_GUARD_LEVEL_ENUM:
                enum_failures.append({
                    "fp_id": fp_id, "marker_number": m.get("marker_number"),
                    "reason": "guard_level_not_in_enum",
                    "value": m.get("guard_level"),
                })
            if not (m.get("guard_reason") or "").strip():
                enum_failures.append({
                    "fp_id": fp_id, "marker_number": m.get("marker_number"),
                    "reason": "empty_guard_reason",
                    "guard_level": m.get("guard_level"),
                })
            if m.get("source_truth") != W17E_SOURCE_TRUTH:
                enum_failures.append({
                    "fp_id": fp_id, "marker_number": m.get("marker_number"),
                    "reason": "source_truth_not_w15e_candidate",
                    "value": m.get("source_truth"),
                })
            if m.get("observation_source") != W17E_OBSERVATION_SOURCE:
                enum_failures.append({
                    "fp_id": fp_id, "marker_number": m.get("marker_number"),
                    "reason": "observation_source_not_w17d_synthesis",
                    "value": m.get("observation_source"),
                })
    inv["guard_level_enum_and_required_reason_present"] = {
        "pass": not enum_failures,
        "detail": {
            "failures": enum_failures[:30],
            "failure_count": len(enum_failures),
            "allowed_levels": list(W17E_GUARD_LEVEL_ENUM),
        },
    }

    # blocking_guard_matches_missing_or_conflict_cases
    blocking_match_failures: List[dict] = []
    for fp_id, payload in payload_by_fp.items():
        mismatch = mismatch_by_fp.get(fp_id) or {}
        vlm_conflict_numbers = {
            c.get("marker_number") for c in
            mismatch.get("vlm_reported_conflicts") or []
            if isinstance(c, dict) and isinstance(c.get("marker_number"), int)
        }
        priorities = priority_by_fp.get(fp_id, {})
        for m in payload.get("markers") or []:
            n = m.get("marker_number")
            bucket = m.get("bucket")
            visible = bool(m.get("visible"))
            level = m.get("guard_level")
            priority = priorities.get(n, "")
            must_be_blocking = (
                bucket in ("missing", "unaccounted_in_readback")
                or not visible
                or (priority == "must_show" and bucket != "matched")
                or (n in vlm_conflict_numbers)
            )
            if must_be_blocking and level != "BLOCKING":
                blocking_match_failures.append({
                    "fp_id": fp_id, "marker_number": n,
                    "reason": "must_be_blocking_but_classified_lower",
                    "bucket": bucket, "visible": visible,
                    "priority": priority,
                    "vlm_conflict": n in vlm_conflict_numbers,
                    "actual_level": level,
                })
            if (level == "BLOCKING") and not must_be_blocking:
                blocking_match_failures.append({
                    "fp_id": fp_id, "marker_number": n,
                    "reason": "blocking_without_missing_or_conflict_basis",
                    "bucket": bucket, "visible": visible,
                    "priority": priority,
                    "vlm_conflict": n in vlm_conflict_numbers,
                })
    inv["blocking_guard_matches_missing_or_conflict_cases"] = {
        "pass": not blocking_match_failures,
        "detail": {
            "failures": blocking_match_failures[:30],
            "failure_count": len(blocking_match_failures),
        },
    }

    inv["production_diff_zero_db_write_zero_vlm_zero_image_zero"] = {
        "pass": (
            production_diff_empty
            and db_write_count == 0
            and not image_import_seen
            and image_api_call_count == 0
            and vlm_api_call_count == 0
            and llm_api_call_count == 0
        ),
        "detail": {
            "production_diff_empty": production_diff_empty,
            "db_write_count": db_write_count,
            "image_import_seen": image_import_seen,
            "image_api_call_count": image_api_call_count,
            "vlm_api_call_count": vlm_api_call_count,
            "llm_api_call_count": llm_api_call_count,
        },
    }

    all_pass = all(v["pass"] for v in inv.values())
    return {"invariants": inv, "all_pass": all_pass}


# ─────────────────────────────────────────────────────────────────────────────
# main
# ─────────────────────────────────────────────────────────────────────────────

def main(argv=None) -> int:
    args = _parse_args(argv)
    run_id = _run_id()
    out_root = Path(args.output_root)
    run_dir = out_root / run_id
    run_dir.mkdir(parents=True, exist_ok=True)

    prev_run_dir = Path(args.derive_payload_from)
    if not prev_run_dir.is_absolute():
        prev_run_dir = Path.cwd() / prev_run_dir

    target_fp_ids, invalid_targets = _resolve_targets(args.target_fp_ids)
    failed_invariants: List[str] = []
    run_status = "succeeded"
    exit_code = 0

    run_meta: Dict[str, Any] = {
        "run_id": run_id,
        "stage": W17E_STAGE,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "args": vars(args),
        "derived_from": prev_run_dir.name,
        "target_fp_ids": sorted(target_fp_ids),
        "invalid_targets": invalid_targets,
        "vlm_api_call_count": 0,
        "image_api_call_count": 0,
        "llm_api_call_count": 0,
        "outputs": [],
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed_invariants,
    }

    def _persist_and_exit(code: int) -> int:
        run_meta["exit_code"] = code
        if code != 0:
            run_meta["run_status"] = "validation_failed"
        run_meta["failed_invariants"] = failed_invariants
        (run_dir / "run_meta.json").write_text(
            json.dumps(run_meta, ensure_ascii=False, indent=2)
        )
        return code

    missing: List[str] = []
    synth_path = prev_run_dir / "coarse_layout_synthesis.json"
    mismatch_path = prev_run_dir / "mismatch_report.json"
    w17d_meta_path = prev_run_dir / "run_meta.json"
    if not synth_path.exists():
        missing.append("coarse_layout_synthesis.json")
    if not mismatch_path.exists():
        missing.append("mismatch_report.json")
    if not w17d_meta_path.exists():
        missing.append("run_meta.json")
    if missing:
        failed_invariants.append("w17d_inputs_missing")
        run_meta["missing_inputs"] = missing
        return _persist_and_exit(1)
    if invalid_targets:
        failed_invariants.append("invalid_target_fp_ids")
        return _persist_and_exit(1)
    if not target_fp_ids.issubset(W16_ALLOWED_TARGET_FP_IDS):
        failed_invariants.append("target_fp_outside_allowed")
        return _persist_and_exit(1)

    synthesis = json.loads(synth_path.read_text())
    mismatch = json.loads(mismatch_path.read_text())
    synthesis_by_fp = synthesis.get("synthesis_by_fp") or {}
    mismatch_by_fp = mismatch.get("mismatch_by_fp") or {}

    priority_by_fp = _load_w17a_legend_priority_index(prev_run_dir)

    payload_by_fp: Dict[str, Dict[str, Any]] = {}
    consumption_by_fp: Dict[str, Dict[str, Any]] = {}
    for fp_id in sorted(target_fp_ids):
        synth_entry = synthesis_by_fp.get(fp_id)
        mismatch_entry = mismatch_by_fp.get(fp_id) or {}
        priority_by_marker = priority_by_fp.get(fp_id, {})
        if not isinstance(synth_entry, dict):
            failed_invariants.append("synthesis_missing_for_target_fp")
            run_meta["missing_fp_id"] = fp_id
            return _persist_and_exit(1)
        payload_entry, consumption_trace = _build_payload_for_fp(
            fp_id=fp_id,
            synthesis_entry=synth_entry,
            mismatch_entry=mismatch_entry,
            priority_by_marker=priority_by_marker,
        )
        payload_by_fp[fp_id] = payload_entry
        consumption_by_fp[fp_id] = consumption_trace

    payload = {
        "payload_by_fp": payload_by_fp,
        "source_truth": W17E_SOURCE_TRUTH,
        "observation_source": W17E_OBSERVATION_SOURCE,
        "guard_level_enum": list(W17E_GUARD_LEVEL_ENUM),
        "policy_note": (
            "downstream payload preview is a deterministic policy "
            "preview, not a semantic proof. W15e candidate stays the "
            "truth surface; W17D synthesis is an observation layer."
        ),
    }
    consumption = {
        "consumption_by_fp": consumption_by_fp,
        "policy_note": payload["policy_note"],
    }

    (run_dir / "downstream_floor_plan_payload_preview.json").write_text(
        json.dumps(payload, ensure_ascii=False, indent=2)
    )
    run_meta["outputs"].append("downstream_floor_plan_payload_preview.json")
    (run_dir / "downstream_guard_consumption_dry_run.json").write_text(
        json.dumps(consumption, ensure_ascii=False, indent=2)
    )
    run_meta["outputs"].append("downstream_guard_consumption_dry_run.json")

    report = _build_w17e_compatibility_report(
        payload_by_fp=payload_by_fp,
        mismatch_by_fp=mismatch_by_fp,
        synthesis_by_fp=synthesis_by_fp,
        priority_by_fp=priority_by_fp,
        production_diff_empty=_check_production_diff_empty(),
        db_write_count=0,
        image_import_seen=_check_image_imports_present(),
        image_api_call_count=0,
        vlm_api_call_count=0,
        llm_api_call_count=0,
        missing_inputs=missing,
        prev_run_id=prev_run_dir.name,
        target_fp_ids=target_fp_ids,
    )
    (run_dir / "w17e_compatibility_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    run_meta["outputs"].append("w17e_compatibility_report.json")

    for name, v in report["invariants"].items():
        if not v["pass"] and name not in failed_invariants:
            failed_invariants.append(name)
    if failed_invariants and run_status == "succeeded":
        run_status = "validation_failed"
        exit_code = 1

    run_meta["run_status"] = run_status
    run_meta["exit_code"] = exit_code
    run_meta["failed_invariants"] = failed_invariants

    # Light HTML.
    def esc(x):
        return (
            str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
        )

    fp_sections: List[str] = []
    for fp_id, payload_entry in payload_by_fp.items():
        marker_rows = "".join(
            f"<tr><td>#{esc(m.get('marker_number'))}</td>"
            f"<td>{esc(m.get('label'))}</td>"
            f"<td>{esc(m.get('element_kind'))}</td>"
            f"<td>{esc(m.get('expected_unit'))}</td>"
            f"<td>{esc(m.get('observed_unit_label_or_area'))}</td>"
            f"<td>{esc(m.get('approximate_region_or_cell'))}</td>"
            f"<td>{esc(m.get('confidence'))}</td>"
            f"<td>{esc(m.get('bucket'))}</td>"
            f"<td>{esc(m.get('priority'))}</td>"
            f"<td><b>{esc(m.get('guard_level'))}</b></td>"
            f"<td>{esc(m.get('guard_reason'))}</td></tr>"
            for m in payload_entry.get("markers") or []
        )
        counts = payload_entry.get("guard_level_counts") or {}
        cons = consumption_by_fp.get(fp_id) or {}
        fp_sections.append(
            f"<section><h2>fp: {esc(fp_id)}</h2>"
            f"<p>guard_level_counts: <pre>{esc(json.dumps(counts))}</pre></p>"
            f"<p>consumer_decision_dry_run: <b>{esc(cons.get('consumer_decision_dry_run'))}</b></p>"
            f"<table><tr><th>#</th><th>label</th><th>kind</th>"
            f"<th>expected_unit</th><th>observed_unit</th><th>region</th>"
            f"<th>confidence</th><th>bucket</th><th>priority</th>"
            f"<th>guard_level</th><th>guard_reason</th></tr>{marker_rows}"
            f"</table></section>"
        )

    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td>"
        f"<td class=\"{'pass' if v['pass'] else 'fail'}\">"
        f"{'PASS' if v['pass'] else 'FAIL'}</td>"
        f"<td><pre>{esc(json.dumps(v.get('detail'), ensure_ascii=False))[:600]}</pre></td>"
        f"</tr>"
        for k, v in (report.get("invariants") or {}).items()
    )
    (run_dir / "index.html").write_text(
        f"""<!doctype html><html><head><meta charset="utf-8">
<title>W17E downstream_payload_preview {esc(run_id)}</title>
<style>body{{font-family:sans-serif;margin:1.5em}}
table{{border-collapse:collapse;margin:0.5em 0}}
td,th{{border:1px solid #ccc;padding:4px 8px;vertical-align:top}}
.pass{{color:#080}} .fail{{color:#b00}}
pre{{white-space:pre-wrap;font-size:0.85em;max-width:84ch}}</style></head><body>
<h1>W17E — downstream_payload_preview {esc(run_id)}</h1>
<p>derived_from(W17D): {esc(prev_run_dir.name)}
| run_status: <b>{esc(run_status)}</b>
| exit_code: {esc(exit_code)}
| vlm_api_call_count: <b>0</b>
| image_api_call_count: <b>0</b>
| llm_api_call_count: <b>0</b></p>

<p>policy_note: <i>{esc(payload['policy_note'])}</i></p>

{''.join(fp_sections)}

<section><h2>invariants</h2>
<table><tr><th>invariant</th><th>status</th><th>detail</th></tr>
{inv_rows}</table></section>

<details><summary>raw run_meta.json</summary>
<pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
</body></html>"""
    )
    run_meta["outputs"].append("index.html")

    (run_dir / "run_meta.json").write_text(
        json.dumps(run_meta, ensure_ascii=False, indent=2)
    )
    _maybe_print_imports(args)
    return exit_code


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