"""Unit tests for build_dispatch_payload + _resolve_start_frame."""

from __future__ import annotations

import json
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

_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 import dispatch_payload as dp  # noqa: E402
from recoil.pipeline._lib.dispatch_payload import (  # noqa: E402
    PayloadContext,
    build_dispatch_payload,
    build_unified_payload,
    DispatchPayloadError,
    DEFAULT_VIDEO_MODEL,
    PromptTooLongError,
    _resolve_start_frame,
)
from recoil.pipeline._lib.grouping import (  # noqa: E402
    ONER_PROMPT_DIRECTIVE,
    GroupingContext,
    get_grouping,
)
from recoil.pipeline._lib.plan_loader import CanonicalShot, CharacterEntry  # noqa: E402


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


@pytest.fixture(autouse=True)
def _clear_project_config_cache():
    """Reset module-global cache between tests so monkeypatching
    load_project_config doesn't poison subsequent tests."""
    from recoil.pipeline._lib import dispatch_payload as _dp

    _dp._project_config_cache.clear()
    yield
    _dp._project_config_cache.clear()


def _shot(
    shot_id="EP001_SH01",
    video_model=None,
    duration_s=3.0,
    **raw_extra,
) -> CanonicalShot:
    raw = {
        "shot_id": shot_id,
        "scene_index": 1,
        "routing_data": {
            "target_editorial_duration_s": 3,
            "is_env_only": True,
            # Satisfies seeddance-2.0 i2v builder pre-flight check
            # (builder only tests truthiness, no file I/O at build time).
            "start_frame_url": "https://fixture.test/fake_start.png",
        },
        "asset_data": {"location_id": "int_lower_decks_corridor", "characters": []},
        "compiled_prompts": {"seeddance_t2v": "test seeddance prompt"},
        "prompt_data": {"shot_type": "WS"},
    }
    raw.update(raw_extra)
    return CanonicalShot(
        shot_id=shot_id,
        scene_index=1,
        sequence_id=None,
        pipeline="still",
        previs_model="gemini-3-pro-image-preview",
        video_model=video_model,
        location_id="int_lower_decks_corridor",
        characters=[],
        shot_type="WS",
        duration_s=duration_s,
        is_env_only=True,
        has_dialogue=False,
        aspect_ratio=None,
        raw=raw,
    )


@pytest.fixture
def fake_start_frame(tmp_path, monkeypatch):
    p = tmp_path / "fake_start.png"
    p.write_bytes(PNG_BYTES)
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload._resolve_start_frame",
        lambda *, shot, project, override: p,
    )
    return p


def test_unknown_modality_raises():
    with pytest.raises(DispatchPayloadError):
        build_dispatch_payload(
            shot=_shot(),
            project="tartarus",
            modality="audio_t2a",
        )


def test_non_canonical_shot_raises(fake_start_frame):
    with pytest.raises(DispatchPayloadError):
        build_dispatch_payload(
            shot={"shot_id": "X"},  # type: ignore[arg-type]
            project="tartarus",
            modality="video_i2v",
        )


def test_model_resolution_order(fake_start_frame):
    # 1. model_override beats everything
    p = build_dispatch_payload(
        shot=_shot(video_model="seeddance-2.0"),
        project="tartarus",
        modality="video_i2v",
        model_override="kling-v3",
        episode="ep_001",
    )
    assert p["model"] == "kling-v3"
    # 2. shot.video_model used when no override
    p = build_dispatch_payload(
        shot=_shot(video_model="seeddance-2.0"),
        project="tartarus",
        modality="video_i2v",
        episode="ep_001",
    )
    assert p["model"] == "seeddance-2.0"
    # 3. DEFAULT_VIDEO_MODEL otherwise
    p = build_dispatch_payload(
        shot=_shot(video_model=None),
        project="tartarus",
        modality="video_i2v",
        episode="ep_001",
    )
    assert p["model"] == DEFAULT_VIDEO_MODEL
    # previs_model is never promoted — covered by the DEFAULT_VIDEO_MODEL assert above
    assert p["model"] != "gemini-3-pro-image-preview"


def test_start_frame_in_payload(fake_start_frame):
    # v2 emits the resolved start frame as a string PATH under "start_frame"
    # (rehydrated downstream), not as a base64 blob under the v1 "image" key.
    p = build_dispatch_payload(
        shot=_shot(),
        project="tartarus",
        modality="video_i2v",
        episode="ep_001",
    )
    assert "start_frame" in p
    assert p["start_frame"] == str(fake_start_frame)


def test_start_frame_resolved_but_file_missing(tmp_path, monkeypatch):
    ghost = tmp_path / "does_not_exist.png"

    def _raise(*, shot, project, override):
        raise FileNotFoundError(f"state says approved but file missing: {ghost}")

    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload._resolve_start_frame",
        _raise,
    )
    with pytest.raises(FileNotFoundError):
        build_dispatch_payload(
            shot=_shot(),
            project="tartarus",
            modality="video_i2v",
            episode="ep_001",
        )


def test_resolve_start_frame_no_sidecar_raises(tmp_path, monkeypatch):
    """No sidecar present → hard FileNotFoundError (no glob fallback)."""
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    # _resolve_start_frame resolves the sidecar dir via
    # ProjectPaths.for_project(), which calls recoil.core.paths.projects_root
    # directly — so we must redirect that name too, or the lookup escapes to
    # the real production tree (and finds a real EP001_SH01 sidecar).
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    # Create the project dir but NO sidecar.
    (tmp_path / "tartarus" / "_pipeline" / "state" / "visual" / "shots").mkdir(parents=True)
    # And populate a sequences dir with files matching the legacy convention —
    # if the function falls back to a glob scan, this is the bait.
    bait_dir = tmp_path / "tartarus" / "sequences" / "ep_001"
    bait_dir.mkdir(parents=True)
    (bait_dir / "shot_001_take1.png").write_bytes(b"BAIT")

    shot = _shot(shot_id="EP001_SH01")
    with pytest.raises(FileNotFoundError) as exc:
        _resolve_start_frame(shot=shot, project="tartarus", override=None)
    # The error must be about the SIDECAR (the SSOT) — not about a glob result.
    assert "sidecar" in str(exc.value).lower() or "EP001_SH01.json" in str(exc.value)


def test_resolve_start_frame_sidecar_says_approved_but_file_missing(
    tmp_path,
    monkeypatch,
):
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    sidecar_dir = tmp_path / "tartarus" / "_pipeline" / "state" / "visual" / "shots"
    sidecar_dir.mkdir(parents=True)
    (sidecar_dir / "EP001_SH01.json").write_text(
        json.dumps(
            {
                "gate_results": {"hero_frame": "sequences/ep_001/ghost.png"},
            }
        )
    )
    # NOT creating the actual file at sequences/ep_001/ghost.png
    shot = _shot(shot_id="EP001_SH01")
    with pytest.raises(FileNotFoundError) as exc:
        _resolve_start_frame(shot=shot, project="tartarus", override=None)
    assert "ghost.png" in str(exc.value) or "missing" in str(exc.value).lower()


def test_resolve_start_frame_sidecar_resolves_via_hero_frame(
    tmp_path,
    monkeypatch,
):
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    sidecar_dir = tmp_path / "tartarus" / "_pipeline" / "state" / "visual" / "shots"
    sidecar_dir.mkdir(parents=True)
    seq_dir = tmp_path / "tartarus" / "sequences" / "ep_001"
    seq_dir.mkdir(parents=True)
    target = seq_dir / "approved.png"
    target.write_bytes(PNG_BYTES)
    (sidecar_dir / "EP001_SH01.json").write_text(
        json.dumps(
            {
                "gate_results": {"hero_frame": "sequences/ep_001/approved.png"},
            }
        )
    )
    shot = _shot(shot_id="EP001_SH01")
    out = _resolve_start_frame(shot=shot, project="tartarus", override=None)
    assert out == target.resolve()


def test_resolve_start_frame_falls_back_to_output_path(
    tmp_path,
    monkeypatch,
):
    """When gate_results.hero_frame absent, use output_path."""
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    sidecar_dir = tmp_path / "tartarus" / "_pipeline" / "state" / "visual" / "shots"
    sidecar_dir.mkdir(parents=True)
    seq_dir = tmp_path / "tartarus" / "sequences" / "ep_001"
    seq_dir.mkdir(parents=True)
    target = seq_dir / "approved.png"
    target.write_bytes(PNG_BYTES)
    (sidecar_dir / "EP001_SH01.json").write_text(
        json.dumps(
            {
                "output_path": "sequences/ep_001/approved.png",
            }
        )
    )
    shot = _shot(shot_id="EP001_SH01")
    out = _resolve_start_frame(shot=shot, project="tartarus", override=None)
    assert out == target.resolve()


def test_resolve_start_frame_does_not_scan_glob(tmp_path, monkeypatch):
    """Negative test: glob-based fallback is gone. Bait the previs dir
    with a perfectly-named file the legacy convention scan would have
    found; assert the function still raises FileNotFoundError because
    the SIDECAR is the SSOT."""
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    sidecar_dir = tmp_path / "tartarus" / "_pipeline" / "state" / "visual" / "shots"
    sidecar_dir.mkdir(parents=True)
    bait_dir = tmp_path / "tartarus" / "sequences" / "ep_001"
    bait_dir.mkdir(parents=True)
    (bait_dir / "shot_001_take1.png").write_bytes(b"BAIT")
    (bait_dir / "shot_001_take2.png").write_bytes(b"BAIT")
    # Sidecar exists but lacks an approved-path field.
    (sidecar_dir / "EP001_SH01.json").write_text(
        json.dumps(
            {
                "shot_id": "EP001_SH01",
                "status": "pending_qc",
            }
        )
    )
    shot = _shot(shot_id="EP001_SH01")
    with pytest.raises(FileNotFoundError):
        _resolve_start_frame(shot=shot, project="tartarus", override=None)


def test_r2v_multi_omits_image_uses_refs(monkeypatch):
    def _raise(*a, **k):
        raise AssertionError("should not resolve start_frame for r2v_multi")

    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload._resolve_start_frame",
        _raise,
    )
    shots = [_shot(shot_id=f"EP001_SH{i:02d}") for i in range(3)]
    p = build_dispatch_payload(
        shot=shots[0],
        project="tartarus",
        modality="r2v_multi",
        batch_shots=shots,
        episode="ep_001",
    )
    assert "start_frame" not in p
    assert p["provider_hints"]["r2v_multi"] is True
    assert p["provider_hints"]["segment_count"] == 3


def test_r2v_multi_without_batch_shots_raises(monkeypatch):
    """Defensive contract: r2v_multi REQUIRES batch_shots."""
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload._resolve_start_frame",
        lambda *, shot, project, override: Path("/dev/null"),
    )
    with pytest.raises(DispatchPayloadError) as exc:
        build_dispatch_payload(
            shot=_shot(),
            project="tartarus",
            modality="r2v_multi",
            batch_shots=None,
            episode="ep_001",
        )
    assert "batch_shots" in str(exc.value)
    # And the same when batch_shots is an empty list (not None).
    with pytest.raises(DispatchPayloadError):
        build_dispatch_payload(
            shot=_shot(),
            project="tartarus",
            modality="r2v_multi",
            batch_shots=[],
            episode="ep_001",
        )


def test_grouping_prompt_directive_reaches_provider_bound_payload_for_oner():
    shots = [
        _shot(shot_id="EP001_SH01", duration_s=4.0),
        _shot(shot_id="EP001_SH02", duration_s=4.0),
    ]
    grouping_ctx = GroupingContext(
        project="tartarus",
        episode=1,
        canonical_plan=None,
        selected_coverage_passes=[],
        tier_map={},
        wildcard_override=None,
    )
    group = get_grouping("oner").assemble(shots, grouping_ctx)[0]

    payload = build_unified_payload(
        PayloadContext(
            project="tartarus",
            modality=group.modality,
            shot_id=group.shots[0].shot_id,
            prompt="Base provider prompt.",
            model_id="seeddance-2.0",
            batch_shots=group.shots if group.modality == "r2v_multi" else None,
            reference_image_paths=[],
            shot=group.shots[0],
            prompt_directive=group.prompt_directive,
        )
    )

    assert "Base provider prompt." in payload["prompt"]
    assert ONER_PROMPT_DIRECTIVE in payload["prompt"]
    for line in ONER_PROMPT_DIRECTIVE.splitlines():
        assert line in payload["prompt"]


def test_r2v_multi_collects_refs_across_batch(tmp_path, monkeypatch):
    """Batch-aware ref collection unions characters across all shots."""
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    # Pre-seed the config cache: the synthetic project has no
    # project_config.json, and the author bootstrap now fails LOUD on config
    # load errors (REC-148 rank 20) instead of silently caching {}.
    from recoil.pipeline._lib import dispatch_payload as _dp

    monkeypatch.setitem(_dp._project_config_cache, "tartarus", {})
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload._resolve_start_frame",
        lambda *, shot, project, override: Path("/dev/null"),
    )
    # Stub out the prompt builder — this test only validates ref collection,
    # not prompt content. The real r2v_multi builder expects characters as
    # dicts ({char_id: ...}); our fixture uses plain strings.
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.get_builder",
        lambda model_id, modality: lambda *a, **k: "stub prompt",
    )
    # Seed two character refs under the fake project, v3 layout.
    # resolve_character_refs walks assets/char/<slug>/ and only counts
    # files matching the conformant pattern {subject}_{type}_{variant}_v{NN}.{ext}
    # — so the hero ref is e.g. assets/char/jade/jade_identity_hero_v01.png.
    proj = tmp_path / "tartarus"
    for slug in ("jade", "wren"):
        char_dir = proj / "assets" / "char" / slug
        char_dir.mkdir(parents=True)
        (char_dir / f"{slug}_identity_hero_v01.png").write_bytes(PNG_BYTES)

    head = CanonicalShot(
        shot_id="EP001_SH01",
        scene_index=1,
        sequence_id=None,
        pipeline="still",
        previs_model="gemini-3-pro-image-preview",
        video_model=None,
        location_id="int_lower_decks_corridor",
        characters=["JADE"],
        shot_type="WS",
        duration_s=3.0,
        is_env_only=True,
        has_dialogue=False,
        aspect_ratio=None,
        raw={
            "shot_id": "EP001_SH01",
            "scene_index": 1,
            "routing_data": {"target_editorial_duration_s": 3, "is_env_only": True},
            "asset_data": {
                "location_id": "int_lower_decks_corridor",
                "characters": ["JADE"],
            },
            "compiled_prompts": {"seeddance_t2v": "test prompt"},
            "prompt_data": {"shot_type": "WS"},
        },
    )
    tail = CanonicalShot(
        shot_id="EP001_SH02",
        scene_index=2,
        sequence_id=None,
        pipeline="still",
        previs_model="gemini-3-pro-image-preview",
        video_model=None,
        location_id="int_lower_decks_corridor",
        characters=["WREN"],
        shot_type="MS",
        duration_s=3.0,
        is_env_only=False,
        has_dialogue=False,
        aspect_ratio=None,
        raw={
            "shot_id": "EP001_SH02",
            "scene_index": 2,
            "routing_data": {"target_editorial_duration_s": 3, "is_env_only": False},
            "asset_data": {
                "location_id": "int_lower_decks_corridor",
                "characters": ["WREN"],
            },
            "compiled_prompts": {"seeddance_t2v": "test prompt"},
            "prompt_data": {"shot_type": "MS"},
        },
    )

    p = build_dispatch_payload(
        shot=head,
        project="tartarus",
        modality="r2v_multi",
        batch_shots=[head, tail],
        episode="ep_001",
    )
    refs = p.get("reference_images", [])
    refs_str = "\n".join(refs)
    assert "/char/jade/" in refs_str, f"JADE ref missing; got refs={refs}"
    assert "/char/wren/" in refs_str, f"WREN ref missing; got refs={refs}"


def test_duration_capped_at_profile_max(fake_start_frame, monkeypatch):
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.get_profile",
        lambda mid: {"max_duration_seconds": 5, "supports_negative_prompt": False},
    )
    p = build_dispatch_payload(
        shot=_shot(duration_s=9.0),
        project="tartarus",
        modality="video_i2v",
        episode="ep_001",
    )
    assert p["duration"] == 5


def test_tier_override_lands_in_hints(fake_start_frame):
    p = build_dispatch_payload(
        shot=_shot(),
        project="tartarus",
        modality="video_i2v",
        tier_override="fast_720p",
        episode="ep_001",
    )
    assert p["provider_hints"]["tier"] == "fast_720p"


def test_negative_prompt_only_when_supported(fake_start_frame, monkeypatch):
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.get_profile",
        lambda mid: {"supports_negative_prompt": True},
    )
    neg_shot = _shot(negative_prompt="extra limbs")
    p = build_dispatch_payload(
        shot=neg_shot,
        project="tartarus",
        modality="video_i2v",
        episode="ep_001",
    )
    assert p.get("negative_prompt") == "extra limbs"
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.get_profile",
        lambda mid: {"supports_negative_prompt": False},
    )
    p = build_dispatch_payload(
        shot=neg_shot,
        project="tartarus",
        modality="video_i2v",
        episode="ep_001",
    )
    assert "negative_prompt" not in p


def test_video_i2v_resolves_sidecar_before_prompt_builder(tmp_path, monkeypatch):
    """Regression for the EP001_SH36 stub-payload bug.

    The seeddance i2v prompt builder pre-flights on
    routing_data.start_frame_path. EP001 plans don't populate that
    field — the approved frame lives in the state sidecar. The
    builder must see the sidecar-resolved path BEFORE running.
    """
    from recoil.pipeline._lib.dispatch_payload import build_dispatch_payload
    from recoil.pipeline._lib.plan_loader import CanonicalShot

    # Fake projects_root pointing at tmp_path so the sidecar resolver
    # finds our fixture. Both the file-resolution path (dispatch_payload's
    # imported name) and the sidecar-dir lookup (ProjectPaths.for_project →
    # recoil.core.paths.projects_root) must be redirected, or the resolver
    # escapes to the real production EP001_SH36 sidecar.
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    # Pre-seed the config cache: the synthetic project has no
    # project_config.json, and the author bootstrap now fails LOUD on config
    # load errors (REC-148 rank 20) instead of silently caching {}.
    from recoil.pipeline._lib import dispatch_payload as _dp

    monkeypatch.setitem(_dp._project_config_cache, "tartarus", {})

    proj = tmp_path / "tartarus"
    sidecar_dir = proj / "_pipeline" / "state" / "visual" / "shots"
    sidecar_dir.mkdir(parents=True)
    frame_rel = "sequences/ep_001/shot_036_take6.png"
    frame_abs = proj / frame_rel
    frame_abs.parent.mkdir(parents=True)
    # 1x1 PNG bytes (any non-empty bytes object works for the encode).
    frame_abs.write_bytes(
        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
        b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01"
        b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
    )
    (sidecar_dir / "EP001_SH36.json").write_text(
        '{"shot_id": "EP001_SH36", '
        '"gate_results": {"hero_frame": "' + frame_rel + '"}, '
        '"output_path": "' + frame_rel + '"}'
    )

    # Canonical shot with NO routing_data.start_frame_path
    # (matches the EP001 plan shape exactly).
    shot = CanonicalShot(
        shot_id="EP001_SH36",
        scene_index=36,
        sequence_id=None,
        pipeline="video",
        previs_model="gemini-3.1-flash-image-preview",
        video_model="seeddance-2.0",
        location_id="shaft_interior",
        characters=["WREN", "JADE"],
        shot_type="WS",
        duration_s=4.0,
        is_env_only=False,
        has_dialogue=False,
        aspect_ratio="9:16",
        raw={
            "shot_id": "EP001_SH36",
            "refless": True,
            "routing_data": {"target_editorial_duration_s": 4},
            "asset_data": {},
            "kinetic_action": "the cables snap and sparks fly",
            "shot_type": "WS",
            "camera_movement": "static",
            "lighting": {},
        },
    )

    payload = build_dispatch_payload(
        shot=shot,
        project="tartarus",
        modality="video_i2v",
        model_override="seeddance-2.0",
    )

    # The prompt builder must have run successfully (no fallback).
    assert "prompt" in payload and payload["prompt"], (
        "Prompt builder produced empty output — sidecar not reaching builder."
    )
    assert "@Image1" in payload["prompt"], (
        "seeddance i2v prompt should anchor on @Image1; got: " + payload["prompt"]
    )
    # v2 ships the resolved start frame as a string PATH under "start_frame"
    # (the v1 "image" base64 key is gone).
    assert payload["start_frame"], "payload['start_frame'] (start frame path) is empty"


def _cap_r2v_batch() -> list[CanonicalShot]:
    def make(shot_id: str, char_id: str, duration_s: float) -> CanonicalShot:
        name = char_id.title()
        raw = {
            "shot_id": shot_id,
            "scene_index": 1,
            "shot_type": "OTS",
            "camera_side": "A",
            "screen_direction": "left-to-right",
            "routing_data": {
                "target_editorial_duration_s": duration_s,
                "has_dialogue": True,
            },
            "prompt_data": {
                "shot_type": "OTS",
                "action_line": f"{name} holds position.",
            },
            "asset_data": {"characters": [{"char_id": char_id}]},
            "audio_data": {
                "dialogue": [{"speaker": char_id, "text": "Hold the line."}]
            },
        }
        return CanonicalShot(
            shot_id=shot_id,
            scene_index=1,
            sequence_id=None,
            pipeline="video",
            previs_model=None,
            video_model="seeddance-2.0",
            location_id=None,
            characters=[CharacterEntry(char_id=char_id)],
            shot_type="OTS",
            duration_s=duration_s,
            is_env_only=False,
            has_dialogue=True,
            aspect_ratio="9:16",
            raw=raw,
        )

    return [
        make("EP001_CAP_SH01", "JADE", 3.0),
        make("EP001_CAP_SH02", "WREN", 3.0),
    ]


def _cap_r2v_ctx(**overrides) -> PayloadContext:
    shots = _cap_r2v_batch()
    values = {
        "project": "tartarus",
        "modality": "r2v_multi",
        "shot_id": "EP001_CAP_PASS",
        "shot": shots[0],
        "batch_shots": shots,
        "model_id": "seeddance-2.0",
        "bible": {},
    }
    values.update(overrides)
    return PayloadContext(**values)


def _patch_provider_cap(monkeypatch, cap: int | None):
    calls = []

    def fake_resolve_adapter(model_id, payload, tier=None):
        calls.append((model_id, payload, tier))
        provider_id = "fal" if cap is None else "flora"
        adapter = SimpleNamespace(max_prompt_chars=cap, provider_id=provider_id)
        return adapter, tier or "default"

    monkeypatch.setattr(
        "recoil.execution.providers.registry.resolve_adapter",
        fake_resolve_adapter,
    )
    return calls


def _patch_cap_prompt_author(
    monkeypatch,
    outputs: list[str],
    *,
    style_suffix: str = "",
):
    calls: list[dict] = []

    def fake_author_pass(_primitive, _strategy, **kwargs):
        calls.append(kwargs)
        idx = min(len(calls) - 1, len(outputs) - 1)
        return outputs[idx]

    monkeypatch.setattr(dp, "load_project_config", lambda _project: {})
    monkeypatch.setattr(
        dp,
        "_collect_reference_images",
        lambda *args, **kwargs: (
            ["/tmp/jade.png", "/tmp/wren.png"],
            {"identity_1": 1, "identity_2": 2},
        ),
    )
    monkeypatch.setattr(
        dp,
        "resolve_strategy",
        lambda *args, **kwargs: (
            "r2v_multi",
            SimpleNamespace(name="directed_prose", required_inputs=[]),
        ),
    )
    monkeypatch.setattr(dp, "author_pass", fake_author_pass)
    monkeypatch.setattr(dp, "verify_authored_prose", lambda *args, **kwargs: [])
    monkeypatch.setattr(
        dp,
        "bind_named_prose",
        lambda authored, *args, **kwargs: SimpleNamespace(
            text=authored,
            payload_refs={},
        ),
    )
    monkeypatch.setattr(
        dp,
        "_append_directed_prose_cinema_style",
        lambda prompt, **kwargs: f"{prompt.rstrip()}{style_suffix}",
    )
    return calls


def test_provider_cap_none_passes_through_untrimmed(monkeypatch):
    _patch_provider_cap(monkeypatch, None)
    authored = (
        "[0:00-0:03] Jade holds the sealed hatch.\n"
        "[0:03-0:06] Wren answers from the smoke. "
        + "untrimmed fal prose "
        + ("x" * 2600)
    )
    _patch_cap_prompt_author(monkeypatch, [authored])

    payload = build_unified_payload(_cap_r2v_ctx())

    assert len(authored) > 2500
    assert payload["prompt"] == authored


def test_provider_cap_2500_enforced(monkeypatch):
    adapter_cap = 2500
    _patch_provider_cap(monkeypatch, adapter_cap)
    ctx = _cap_r2v_ctx(
        prompt="x" * (adapter_cap + 1),
        reference_image_paths=["/tmp/ref.png"],
    )

    with pytest.raises(PromptTooLongError):
        build_unified_payload(ctx)


def test_resolve_provider_cap_projection_matches_full_payload(monkeypatch):
    from recoil.execution import video_model_client
    from recoil.execution.providers import registry

    adapter_caps = {
        (True, True, True, "720p"): 2468,
        (False, False, False, "720p"): None,
    }
    resolve_calls = []

    def fake_resolve_adapter(model_id, payload, tier=None):
        key = (
            bool(payload.reference_images),
            bool(payload.generate_audio),
            bool(payload.negative_prompt),
            payload.resolution,
        )
        cap = adapter_caps.get(key, 1357)
        resolve_calls.append((model_id, payload, tier, cap))
        return SimpleNamespace(max_prompt_chars=cap, provider_id=f"fake-{cap}"), (
            tier or "default"
        )

    monkeypatch.setattr(registry, "resolve_adapter", fake_resolve_adapter)
    monkeypatch.setattr(
        dp,
        "get_profile",
        lambda _model_id: {
            "supports_audio": True,
            "supports_negative_prompt": True,
            "min_duration_seconds": 1,
            "max_duration_seconds": 15,
        },
    )

    ctx = _cap_r2v_ctx(
        prompt="short prompt",
        reference_image_paths=["/tmp/ref-a.png", "/tmp/ref-b.png"],
        generate_audio=True,
        negative_prompt="extra limbs",
        tier="standard_720p",
    )
    payload = build_unified_payload(ctx)
    unified = video_model_client._dict_to_unified(payload, payload["model"])
    full_adapter, _ = registry.resolve_adapter(
        payload["model"],
        unified,
        tier=payload["provider_hints"]["tier"],
    )
    projection_cap = dp._resolve_provider_cap(
        payload["model"],
        reference_images=payload["reference_images"],
        generate_audio=payload["generate_audio"],
        negative_prompt=payload["negative_prompt"],
        image=payload.get("image"),
        image_tail=payload.get("image_tail"),
        resolution=payload.get("resolution") or "720p",
        tier=payload["provider_hints"]["tier"],
    )

    assert projection_cap == full_adapter.max_prompt_chars == 2468
    assert resolve_calls[-1][1].reference_images == unified.reference_images
    assert resolve_calls[-1][1].generate_audio is unified.generate_audio
    assert resolve_calls[-1][1].negative_prompt == unified.negative_prompt
    assert resolve_calls[-1][1].resolution == unified.resolution


def test_count_after_retry_reauthors_under_cap(monkeypatch):
    adapter_cap = 2500
    _patch_provider_cap(monkeypatch, adapter_cap)
    dialogue = "Hold the line."
    under_cap = (
        f'[0:00-0:03] Jade whispers: "{dialogue}" while bracing the hatch.\n'
        "[0:03-0:06] Wren crosses behind her as the camera tracks the smoke."
    )
    over_cap = under_cap + "\n" + ("tighten this action prose. " * 140)
    style_suffix = "\n\nStyle: locked film grain."
    author_calls = _patch_cap_prompt_author(
        monkeypatch,
        [over_cap, under_cap],
        style_suffix=style_suffix,
    )

    payload = build_unified_payload(_cap_r2v_ctx())
    prompt = payload["prompt"]

    assert len(author_calls) == 2
    assert len(prompt) <= adapter_cap
    assert "Style: locked film grain." in prompt
    assert f'"{dialogue}"' in prompt
    assert "[0:00-0:03]" in prompt
    assert "[0:03-0:06]" in prompt
    assert f"over the {adapter_cap} cap" in author_calls[1][
        "project_config"
    ]["prose_author_retry_failures"]


def test_prompt_too_long_raises_after_max_retries(monkeypatch):
    adapter_cap = 2500
    _patch_provider_cap(monkeypatch, adapter_cap)
    authored = (
        "[0:00-0:03] Jade holds the hatch.\n"
        "[0:03-0:06] Wren answers. "
        + ("x" * adapter_cap)
    )
    author_calls = _patch_cap_prompt_author(monkeypatch, [authored])

    with pytest.raises(PromptTooLongError):
        build_unified_payload(_cap_r2v_ctx())

    assert len(author_calls) == 3


def test_build_unified_payload_fail_loud_over_cap(monkeypatch):
    adapter_cap = 2500
    _patch_provider_cap(monkeypatch, adapter_cap)

    with pytest.raises(PromptTooLongError):
        build_unified_payload(
            _cap_r2v_ctx(
                prompt="x" * (adapter_cap + 1),
                reference_image_paths=["/tmp/ref.png"],
            )
        )


def test_audit_dry_run_payload_fail_loud_over_cap(monkeypatch):
    adapter_cap = 2500
    _patch_provider_cap(monkeypatch, adapter_cap)
    monkeypatch.setattr(dp, "load_project_config", lambda _project: {})
    monkeypatch.setattr(
        dp,
        "_collect_reference_images",
        lambda *args, **kwargs: (["/tmp/ref.png"], {"identity_1": 1}),
    )
    monkeypatch.setattr(
        dp,
        "_build_author_aware_prompt",
        lambda *args, **kwargs: dp.AuthorPromptResult(
            prompt="x" * (adapter_cap + 1),
            modality="r2v_multi",
            strategy="directed_prose",
            payload_refs={},
            fallback=False,
        ),
    )
    shots = _cap_r2v_batch()

    with pytest.raises(PromptTooLongError):
        build_dispatch_payload(
            shot=shots[0],
            project="tartarus",
            modality="r2v_multi",
            batch_shots=shots,
            model_override="seeddance-2.0",
            dry_run=True,
        )


# ============================================================================
# payload_assembly convergence tests (added 2026-05-25)
# SYNTHESIS: consultations/recoil/payload-assembly-convergence-2026-05-25/
# ============================================================================
from unittest import mock  # noqa: E402


_FIXTURES_DIR = (
    Path(__file__).resolve().parents[4]
    / "consultations/recoil/payload-assembly-convergence-2026-05-25/_fixtures"
)


def _make_fixture_shot_i2v() -> CanonicalShot:
    """Recreate the Phase 0 i2v fixture shot exactly."""
    return CanonicalShot(
        shot_id="FIXTURE_I2V_SH01",
        scene_index=0,
        sequence_id="SEQ01",
        pipeline="i2v",
        previs_model=None,
        video_model="seeddance-2.0",
        location_id="int_lobby",
        characters=[CharacterEntry(char_id="HERO")],
        shot_type="medium",
        duration_s=5.0,
        is_env_only=False,
        has_dialogue=False,
        aspect_ratio="9:16",
        raw={"prompt": "test prompt for i2v fixture", "refless": True},
    )


def _make_fixture_shot_t2v() -> CanonicalShot:
    """Recreate the Phase 0 t2v-baseline fixture shot exactly."""
    return CanonicalShot(
        shot_id="FIXTURE_T2V_SH01",
        scene_index=0,
        sequence_id="SEQ01",
        pipeline="t2v",
        previs_model=None,
        video_model="seeddance-2.0",
        location_id=None,
        characters=[],
        shot_type="wide",
        duration_s=5.0,
        is_env_only=True,
        has_dialogue=False,
        aspect_ratio="9:16",
        raw={"prompt": "test prompt for t2v fixture — no characters"},
    )


def _make_fixture_batch() -> list[CanonicalShot]:
    """Recreate the Phase 0 r2v_multi fixture batch exactly."""
    head = _make_fixture_shot_i2v()
    tail = CanonicalShot(
        shot_id="FIXTURE_R2V_SH02",
        scene_index=0,
        sequence_id="SEQ01",
        pipeline="r2v",
        previs_model=None,
        video_model="seeddance-2.0",
        location_id="int_lobby",
        characters=[CharacterEntry(char_id="HERO")],
        shot_type="wide",
        duration_s=4.0,
        is_env_only=False,
        has_dialogue=False,
        aspect_ratio="9:16",
        raw={"prompt": "test prompt 2", "refless": True},
    )
    return [head, tail]


def _normalize_for_parity(payload: dict) -> dict:
    """JSON round-trip to normalize types (Path -> str, etc.) and remove
    insertion-order artifacts."""
    return json.loads(json.dumps(payload, default=str, sort_keys=True))


def test_wrapper_output_parity_i2v():
    """Phase 2 wrapper must produce a payload byte-identical (after JSON
    serialization) to the Phase 0 captured fixture for the video_i2v
    narrative path. Validates SYNTHESIS Condition 1 — wrapper output parity.
    """
    fixture_path = _FIXTURES_DIR / "payload_fixture_i2v.json"
    if not fixture_path.exists():
        pytest.skip(f"Fixture not present at {fixture_path}; Phase 0 must run first")
    expected = _normalize_for_parity(json.loads(fixture_path.read_text()))
    shot = _make_fixture_shot_i2v()
    got = _normalize_for_parity(
        build_dispatch_payload(
            shot=shot,
            project="tartarus",
            modality="video_i2v",
            dry_run=True,
        )
    )
    assert got == expected, (
        "Wrapper output drift (i2v).\n"
        f"  Extra keys: {set(got.keys()) - set(expected.keys())}\n"
        f"  Missing keys: {set(expected.keys()) - set(got.keys())}\n"
        f"  Mismatched fields: {[k for k in set(got.keys()) & set(expected.keys()) if got[k] != expected[k]]}"
    )


def test_wrapper_output_parity_r2v_multi():
    """Wrapper parity for r2v_multi batch path."""
    fixture_path = _FIXTURES_DIR / "payload_fixture_r2v_multi.json"
    if not fixture_path.exists():
        pytest.skip(f"Fixture not present at {fixture_path}")
    expected = _normalize_for_parity(json.loads(fixture_path.read_text()))
    batch = _make_fixture_batch()
    got = _normalize_for_parity(
        build_dispatch_payload(
            shot=batch[0],
            project="tartarus",
            modality="r2v_multi",
            batch_shots=batch,
            dry_run=True,
        )
    )
    assert got == expected, (
        "Wrapper output drift (r2v_multi).\n"
        f"  Extra keys: {set(got.keys()) - set(expected.keys())}\n"
        f"  Missing keys: {set(expected.keys()) - set(got.keys())}\n"
        f"  Mismatched fields: {[k for k in set(got.keys()) & set(expected.keys()) if got[k] != expected[k]]}"
    )


def test_wrapper_output_parity_t2v_baseline():
    """Wrapper parity for the t2v baseline shape."""
    fixture_path = _FIXTURES_DIR / "payload_fixture_t2v_baseline.json"
    if not fixture_path.exists():
        pytest.skip(f"Fixture not present at {fixture_path}")
    expected = _normalize_for_parity(json.loads(fixture_path.read_text()))
    shot = _make_fixture_shot_t2v()
    got = _normalize_for_parity(
        build_dispatch_payload(
            shot=shot,
            project="tartarus",
            modality="video_i2v",
            dry_run=True,
        )
    )
    assert got == expected, (
        "Wrapper output drift (t2v baseline).\n"
        f"  Extra keys: {set(got.keys()) - set(expected.keys())}\n"
        f"  Missing keys: {set(expected.keys()) - set(got.keys())}\n"
        f"  Mismatched fields: {[k for k in set(got.keys()) & set(expected.keys()) if got[k] != expected[k]]}"
    )


def test_dry_run_output_filename_uses_grouping_strategy_and_ordinal():
    batch = _make_fixture_batch()

    payload = build_dispatch_payload(
        shot=batch[0],
        project="tartarus",
        modality="r2v_multi",
        batch_shots=batch,
        episode="ep_001",
        dry_run=True,
        skip_author=True,
        force_no_refs=True,
        grouping={
            "strategy": "continuity",
            "ordinal": 2,
            "shot_ids": [s.shot_id for s in batch],
            "source_pass_id": None,
        },
        generation_config={"tier": "pro", "seed": 1234},
        element_config={"identity_ref_mode": "full_turnaround"},
    )

    assert payload["output_filename"] == "EP001_CONT_002_SH01_02_take1.mp4"
    assert payload["grouping"]["strategy"] == "continuity"
    assert payload["grouping"]["ordinal"] == 2
    assert payload["generation_config"] == {"tier": "pro", "seed": 1234}
    assert payload["provider_hints"]["tier"] == "pro"
    assert payload["provider_hints"]["seed"] == 1234
    assert payload["element_config"] == {"identity_ref_mode": "full_turnaround"}


def test_dry_run_output_filename_omitted_when_r2v_grouping_unknown():
    batch = _make_fixture_batch()

    payload = build_dispatch_payload(
        shot=batch[0],
        project="tartarus",
        modality="r2v_multi",
        batch_shots=batch,
        episode="ep_001",
        dry_run=True,
        skip_author=True,
        force_no_refs=True,
    )

    assert "output_filename" not in payload


def test_cli_override_precedence_start_frame(tmp_path):
    """If PayloadContext.start_frame_path is set, _resolve_start_frame
    is not called. Validates SYNTHESIS Audit Assertion #7 / unit test.

    Also validates the post-build code-review fix: an explicit
    start_frame_path that does not exist on disk raises FileNotFoundError
    regardless of whether ctx.shot is set (legacy parity).
    """
    real_frame = tmp_path / "explicit_start.png"
    real_frame.write_bytes(PNG_BYTES)
    ctx = PayloadContext(
        project="test",
        modality="video_i2v",
        shot_id="TEST_CLI_OVERRIDE",
        prompt="cli prompt",
        model_id="seeddance-2.0",
        start_frame_path=real_frame,
        reference_image_paths=[],
    )
    with mock.patch("recoil.pipeline._lib.dispatch_payload._resolve_start_frame") as m:
        payload = build_unified_payload(ctx)
        m.assert_not_called()
    assert payload["model"] == "seeddance-2.0"
    assert payload["shot_id"] == "TEST_CLI_OVERRIDE"
    assert payload["start_frame"] == str(real_frame)

    # Missing override path → FileNotFoundError even when ctx.shot is None.
    ghost_ctx = PayloadContext(
        project="test",
        modality="video_i2v",
        shot_id="TEST_CLI_GHOST",
        prompt="cli prompt",
        model_id="seeddance-2.0",
        start_frame_path=tmp_path / "does_not_exist.png",
        reference_image_paths=[],
    )
    with pytest.raises(FileNotFoundError):
        build_unified_payload(ghost_ctx)


def _real_png_bytes(w: int = 640, h: int = 640) -> bytes:
    """A genuinely real PNG (random-noise pixels so it clears _MIN_SHEET_BYTES;
    >=512x512) for composite-sheet fixtures — must pass dispatch_payload._sheet_is_real."""
    import os as _os
    from io import BytesIO

    from PIL import Image

    im = Image.frombytes("RGB", (w, h), _os.urandom(w * h * 3))
    buf = BytesIO()
    im.save(buf, "PNG")
    return buf.getvalue()


def test_composite_sheets_env_unset_no_op(tmp_path, monkeypatch):
    """Default (env var unset AND project config not enabling) → sheet helper
    returns None, angle path runs."""
    monkeypatch.delenv("RECOIL_USE_COMPOSITE_SHEETS", raising=False)
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    import recoil.pipeline._lib.dispatch_payload as _dp

    # Disable the project-config channel too (real tartarus config enables sheets),
    # so this test isolates "nothing enables → no-op" rather than reading live config.
    monkeypatch.setitem(_dp._project_config_cache, "tartarus", {})
    from recoil.pipeline._lib.dispatch_payload import _collect_sheet_refs

    shot = _shot(shot_id="EP001_SH02")
    assert _collect_sheet_refs(shot, "tartarus", None) is None


def test_composite_sheets_missing_sheet_falls_back(tmp_path, monkeypatch):
    """REC-34: env=1 (composite ENABLED) but a required sheet is ABSENT → fall back
    to the parked angle path (return None) with a loud warning. Missing ≠ corrupt:
    a missing sheet is a not-yet-cast entity; a present-but-fake sheet is the bug we
    hard-fail on (see the corrupt-sheet test)."""
    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)
    (tmp_path / "tartarus").mkdir(parents=True, exist_ok=True)  # project dir exists; sheet does not
    from recoil.pipeline._lib.dispatch_payload import (
        _collect_sheet_refs,
        CharacterEntry,
    )

    shot = _shot(shot_id="EP001_SH02")
    shot = CanonicalShot(
        **{**shot.__dict__, "characters": [CharacterEntry(char_id="JADE")]}
    )
    assert _collect_sheet_refs(shot, "tartarus", None) is None


def test_composite_sheets_env_set_files_present_returns_sheets(
    tmp_path,
    monkeypatch,
):
    """env=1 AND all required sheets exist → helper returns sheet refs + manifest."""
    monkeypatch.setenv("RECOIL_USE_COMPOSITE_SHEETS", "1")
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    # ProjectPaths.sheet_path resolves the per-kind layout via ProjectPaths.for_project →
    # recoil.core.paths.projects_root; redirect it too so the sheet lookup
    # lands inside tmp_path rather than the real production tree.
    monkeypatch.setattr(
        "recoil.core.paths.projects_root",
        lambda: tmp_path,
    )
    from recoil.pipeline._lib.dispatch_payload import (
        _collect_sheet_refs,
        CharacterEntry,
    )

    # Drop sheets at the canonical paths. ProjectPaths.sheet_path routes
    # through ProjectPaths sheet dir helpers (Phase 2a N3):
    #   characters → get_character_sheets_dir → assets/char/<slug>/base/sheets/
    #   locations  → get_location_sheets_dir  → assets/loc/<slug>/sheets/
    jade_sheet = (
        tmp_path
        / "tartarus"
        / "assets"
        / "char"
        / "jade"
        / "base"
        / "sheets"
        / "sheet_v1.png"
    )
    loc_sheet = (
        tmp_path
        / "tartarus"
        / "assets"
        / "loc"
        / "int_lower_decks_corridor"
        / "sheets"
        / "sheet_v1.png"
    )
    for p in (jade_sheet, loc_sheet):
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_bytes(_real_png_bytes())

    shot = _shot(shot_id="EP001_SH02")
    shot = CanonicalShot(
        **{**shot.__dict__, "characters": [CharacterEntry(char_id="JADE")]}
    )
    result = _collect_sheet_refs(shot, "tartarus", None)
    assert result is not None
    refs, manifest = result
    assert len(refs) == 2  # 1 character + 1 location
    assert any("jade/base/sheets/sheet_v1.png" in r for r in refs)
    assert any("int_lower_decks_corridor/sheets/sheet_v1.png" in r for r in refs)
    assert manifest["identity_1"] == 1
    assert manifest["scene_1"] == 2


def test_composite_sheets_corrupt_sheet_raises_no_silent_t2v(tmp_path, monkeypatch):
    """REC-34 core regression: a sheet that EXISTS but is corrupt/1x1/truncated
    (the REC-31/REC-33 synthetic-clobber class) must LOUD-FAIL the dispatch — never
    pass the old existence-only check and silently ship a ref-less generation."""
    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.dispatch_payload import (
        _collect_sheet_refs,
        CharacterEntry,
    )
    from recoil.core.ref_errors import SheetIntegrityError

    jade_sheet = (
        tmp_path / "tartarus" / "assets" / "char" / "jade" / "base" / "sheets" / "sheet_v1.png"
    )
    jade_sheet.parent.mkdir(parents=True, exist_ok=True)
    jade_sheet.write_bytes(PNG_BYTES)  # exists but <1KB — the clobber class

    shot = _shot(shot_id="EP001_SH02")
    shot = CanonicalShot(
        **{**shot.__dict__, "characters": [CharacterEntry(char_id="JADE")]}
    )
    with pytest.raises(SheetIntegrityError):
        _collect_sheet_refs(shot, "tartarus", None)


def test_composite_sheets_env_set_no_entities_returns_none(tmp_path, monkeypatch):
    """env=1 but shot has no characters AND no location → fall back (angle path
    will handle the env-only case correctly)."""
    monkeypatch.setenv("RECOIL_USE_COMPOSITE_SHEETS", "1")
    monkeypatch.setattr(
        "recoil.pipeline._lib.dispatch_payload.projects_root",
        lambda: tmp_path,
    )
    from recoil.pipeline._lib.dispatch_payload import _collect_sheet_refs

    shot = CanonicalShot(
        shot_id="EP001_ENV",
        scene_index=1,
        sequence_id=None,
        pipeline="still",
        previs_model=None,
        video_model=None,
        location_id=None,
        characters=[],
        shot_type="WS",
        duration_s=3.0,
        is_env_only=True,
        has_dialogue=False,
        aspect_ratio="9:16",
        raw={"shot_id": "EP001_ENV"},
    )
    assert _collect_sheet_refs(shot, "tartarus", None) is None


def test_composite_sheets_project_config_enables_path(tmp_path, monkeypatch):
    """env unset BUT project_config.json::use_composite_sheets=true → enabled.

    Persistent per-project opt-in path (2026-05-28). JT doesn't have to
    remember the env var for every production fire.
    """
    monkeypatch.delenv("RECOIL_USE_COMPOSITE_SHEETS", raising=False)
    from recoil.pipeline._lib import dispatch_payload as _dp

    _dp._project_config_cache.clear()
    monkeypatch.setattr(
        _dp, "load_project_config", lambda project: {"use_composite_sheets": True}
    )
    assert _dp._composite_sheets_enabled("tartarus") is True


def test_composite_sheets_project_config_false_keeps_off(tmp_path, monkeypatch):
    """env unset AND project_config::use_composite_sheets=false → disabled.

    Default-off is preserved when the project hasn't opted in.
    """
    monkeypatch.delenv("RECOIL_USE_COMPOSITE_SHEETS", raising=False)
    from recoil.pipeline._lib import dispatch_payload as _dp

    _dp._project_config_cache.clear()
    monkeypatch.setattr(
        _dp, "load_project_config", lambda project: {"use_composite_sheets": False}
    )
    assert _dp._composite_sheets_enabled("tartarus") is False


def test_composite_sheets_project_config_missing_key_keeps_off(tmp_path, monkeypatch):
    """env unset AND project_config has no use_composite_sheets key → disabled."""
    monkeypatch.delenv("RECOIL_USE_COMPOSITE_SHEETS", raising=False)
    from recoil.pipeline._lib import dispatch_payload as _dp

    _dp._project_config_cache.clear()
    monkeypatch.setattr(
        _dp, "load_project_config", lambda project: {"schema_version": 2}
    )
    assert _dp._composite_sheets_enabled("tartarus") is False


def test_composite_sheets_env_wins_over_project_config(tmp_path, monkeypatch):
    """env=1 always enables, even when project_config explicitly says False.

    Env var is the ad-hoc override channel — wins so a developer can force
    the path on for one fire regardless of project state.
    """
    monkeypatch.setenv("RECOIL_USE_COMPOSITE_SHEETS", "1")
    from recoil.pipeline._lib import dispatch_payload as _dp

    _dp._project_config_cache.clear()
    monkeypatch.setattr(
        _dp, "load_project_config", lambda project: {"use_composite_sheets": False}
    )
    assert _dp._composite_sheets_enabled("tartarus") is True


def test_composite_sheets_config_load_error_falls_back_to_off(monkeypatch):
    """If load_project_config raises, treat as not-opted-in (don't crash)."""
    monkeypatch.delenv("RECOIL_USE_COMPOSITE_SHEETS", raising=False)
    from recoil.pipeline._lib import dispatch_payload as _dp

    _dp._project_config_cache.clear()

    def _raise(project):
        raise RuntimeError("config corrupted")

    monkeypatch.setattr(_dp, "load_project_config", _raise)
    assert _dp._composite_sheets_enabled("tartarus") is False


def test_cli_override_precedence_refs():
    """If PayloadContext.reference_image_paths is set, _collect_reference_images
    is not called.
    """
    ctx = PayloadContext(
        project="test",
        modality="video_wan_r2v",
        shot_id="TEST_CLI_REFS",
        prompt="wan prompt",
        model_id="wan-2.7-r2v",
        reference_image_paths=[Path("/tmp/ref_a.jpg"), Path("/tmp/ref_b.jpg")],
        multi_shots=True,
    )
    with mock.patch(
        "recoil.pipeline._lib.dispatch_payload._collect_reference_images"
    ) as m:
        payload = build_unified_payload(ctx)
        m.assert_not_called()
    assert payload["reference_images"] == ["/tmp/ref_a.jpg", "/tmp/ref_b.jpg"]
    assert payload["provider_hints"]["multi_shots"] is True