"""REC-213 C1+C2 — PIPELINE live-path test: the r2v collector reaches composite
SHEETS via the bundle (resolve_sheet_asset), multi-entity, with the unchanged
fail-closed gate satisfied by staged heroes.

Lives under pipeline/_lib/tests (NOT core/tests) because it exercises
recoil.pipeline._lib.dispatch_payload._collect_reference_images. Staging mirrors
test_rec31_gate_on_collector._stage_two_shelf_heroes."""
from __future__ import annotations

import logging
import os
import sys
from pathlib import Path

import pytest
from PIL import Image

_REPO_ROOT = Path(__file__).resolve().parents[4]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from recoil.pipeline._lib.dispatch_payload import _collect_reference_images  # noqa: E402
from recoil.pipeline._lib.plan_loader import CanonicalShot, CharacterEntry  # noqa: E402
from recoil.core.paths import ProjectPaths  # noqa: E402
from recoil.core.ref_errors import SheetIntegrityError  # noqa: E402

PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
_LOC = "int_lower_decks_maintenance_shaft"


def _real_sheet(p: Path) -> None:
    """Valid sheet: PNG magic, 1024x1024, noise so it clears the 100KB floor."""
    p.parent.mkdir(parents=True, exist_ok=True)
    Image.frombytes("RGB", (1024, 1024), os.urandom(1024 * 1024 * 3)).save(p, "PNG")


def _stage(tmp_path, monkeypatch, *, chars=("WREN", "JADE"), sheets=None):
    """Shelf-hero char(s) + location, composite sheets ENABLED. `chars` is the
    shot roster (each gets a shelf hero so the unchanged gate passes); `sheets`
    (default: all chars + location) controls which entities get a VALID sheet."""
    monkeypatch.setenv("RECOIL_USE_COMPOSITE_SHEETS", "1")
    monkeypatch.setattr("recoil.pipeline._lib.dispatch_payload.projects_root", lambda: tmp_path)
    monkeypatch.setattr("recoil.core.paths.projects_root", lambda: tmp_path)
    from recoil.pipeline._lib import dispatch_payload as _dp
    monkeypatch.setitem(_dp._project_config_cache, "tartarus", {})

    proj = tmp_path / "tartarus"
    for cid in chars:
        look = proj / "assets" / "char" / cid.lower() / "base"
        look.mkdir(parents=True)
        (look / f"{cid.lower()}-identity.png").write_bytes(PNG_BYTES)  # shelf hero -> gate passes

    pp = ProjectPaths.for_project("tartarus")
    paths = {cid: pp.sheet_path("char", cid) for cid in chars}
    paths[_LOC] = pp.sheet_path("loc", _LOC)
    want = sheets if sheets is not None else (*chars, _LOC)
    for key in want:
        _real_sheet(paths[key])

    raw = {"shot_id": "EP001_SH37", "scene_index": 1,
           "asset_data": {"location_id": _LOC,
                          "characters": [{"char_id": c} for c in chars]},
           "compiled_prompts": {"seeddance_t2v": "p"}, "prompt_data": {"shot_type": "MS"}}
    cs = CanonicalShot(
        shot_id="EP001_SH37", scene_index=1, sequence_id=None, pipeline="r2v",
        previs_model="gemini-3-pro-image-preview", video_model="seedream-v4.5",
        location_id=_LOC,
        characters=[CharacterEntry(char_id=c) for c in chars],
        shot_type="MS", duration_s=3.0, is_env_only=False, has_dialogue=False,
        aspect_ratio=None, raw=raw,
    )
    return cs, paths


def test_video_payload_sheet_parity_multi(tmp_path, monkeypatch):
    """MULTI-entity: 2 chars + location, all with sheets -> 3 sheet refs, manifest
    {identity_1, identity_2, scene_1}. Proves per-entity concatenation does NOT
    collapse to a single sheet."""
    cs, paths = _stage(tmp_path, monkeypatch)
    refs, manifest = _collect_reference_images(cs, "tartarus", "r2v_multi", batch_shots=[cs])
    assert manifest == {"identity_1": 1, "identity_2": 2, "scene_1": 3}, manifest
    assert len(refs) == 3, refs
    # roster order: WREN, JADE, then location
    assert refs[0] == str(paths["WREN"])
    assert refs[1] == str(paths["JADE"])
    assert refs[2] == str(paths[_LOC])
    assert all("sheet_v1.png" in r for r in refs)


def test_video_payload_sheet_parity_one_char(tmp_path, monkeypatch):
    """ONE char + location (spec-required parity case) -> 2 sheet refs, manifest
    {identity_1, scene_1} — matches what the deleted _try_composite_sheet_refs produced."""
    cs, paths = _stage(tmp_path, monkeypatch, chars=("JADE",))
    refs, manifest = _collect_reference_images(cs, "tartarus", "r2v_multi", batch_shots=[cs])
    assert manifest == {"identity_1": 1, "scene_1": 2}, manifest
    assert refs == [str(paths["JADE"]), str(paths[_LOC])], refs


def test_all_or_nothing_missing_sheet_falls_back_with_warning(tmp_path, monkeypatch, caplog):
    """One required sheet ABSENT -> the WHOLE shot falls back to angle-based refs
    (NOT sheets), and a WARNING naming the missing entity fires (anti-masking)."""
    cs, paths = _stage(tmp_path, monkeypatch, sheets=("WREN", "JADE"))  # location sheet missing
    with caplog.at_level(logging.WARNING):
        refs, manifest = _collect_reference_images(cs, "tartarus", "r2v_multi", batch_shots=[cs])
    # fell back to angle path: shelf heroes, NOT composite sheets
    assert not any("sheet_v1.png" in r for r in refs), refs
    assert any("wren-identity.png" in r for r in refs)
    # warning must name BOTH the missing entity AND the canonical path that was absent
    loc_sheet = str(paths[_LOC])
    assert any("composite sheet MISSING" in rec.getMessage() and _LOC in rec.getMessage()
               and loc_sheet in rec.getMessage() for rec in caplog.records), caplog.text


def test_corrupt_sheet_raises(tmp_path, monkeypatch):
    """A present-but-corrupt sheet -> SheetIntegrityError (loud abort, no spend)."""
    cs, paths = _stage(tmp_path, monkeypatch)
    paths[_LOC].write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)  # truncated clobber
    with pytest.raises(SheetIntegrityError):
        _collect_reference_images(cs, "tartarus", "r2v_multi", batch_shots=[cs])