from __future__ import annotations

import asyncio
import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

from recoil.execution.step_types import ProjectPaths
from recoil.pipeline._lib.plan_loader import CanonicalPlan, CanonicalShot
from recoil.pipeline.cli import generate
from recoil.pipeline.core.persistence import (
    SceneIdentityMismatchError,
    load_manifest,
    save_scene,
    scene_manifest_path,
    scene_path,
    scene_version_path,
)
from recoil.pipeline.core.scene_version_store import SceneVersionStore
from recoil.pipeline.core.take import Beat, Scene
from recoil.pipeline.orchestrator.episode_runner import EpisodeRunner
from recoil.workspace import readmodel


PROJECT = "fixture"
BATCH_ID = "BATCH_004"

# ── REC-231 Phase 6: operator loop + identity-halt + kind ─────────────────────────
EP = "ep_001"
BATCH = "BATCH_001"
SELECTOR = "EP001_CONT_001"
_GROUPING = {"strategy": "continuity", "ordinal": 1}
_ART_V1 = "prep/ep_001/storyboards/BATCH_001_v1.png"
_ART_V2 = "prep/ep_001/storyboards/BATCH_001_v2.png"


@pytest.fixture(autouse=True)
def _projects_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    root = tmp_path / "projects"
    root.mkdir()
    (root / ".recoil-data-root").touch()
    project_root = root / PROJECT
    project_root.mkdir()
    monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(root))
    return project_root


def _shot(batch_index: int, shot_index: int) -> dict:
    ordinal = (batch_index - 1) * 3 + shot_index
    return {
        "shot_id": f"EP001_SH{ordinal:02d}",
        "scene_index": batch_index,
        "pipeline": "video",
        "video_model": "seeddance-2.0",
        "asset_data": {
            "characters": [],
            "location_id": f"LOC_{batch_index}",
        },
        "prompt_data": {"shot_type": "MS"},
        "routing_data": {
            "target_editorial_duration_s": 2.0,
            "is_env_only": False,
            "has_dialogue": False,
        },
        "aspect_ratio": "9:16",
    }


def _write_plan() -> Path:
    paths = ProjectPaths.for_episode(PROJECT, 1)
    paths.plans_dir.mkdir(parents=True, exist_ok=True)
    plan_path = paths.plans_dir / "ep_001_plan.json"
    shots = [
        _shot(batch_index, shot_index)
        for batch_index in range(1, 5)
        for shot_index in range(1, 4)
    ]
    plan_path.write_text(
        json.dumps(
            {
                "episode_id": "ep_001",
                "project": PROJECT,
                "shots": shots,
            },
            indent=2,
        ),
        encoding="utf-8",
    )
    return plan_path


def _save_locked_scene() -> Path:
    scene = Scene(
        scene_id=BATCH_ID,
        beats=[
            Beat(
                beat_id=BATCH_ID,
                beat_metadata={"scene_id": BATCH_ID, "marker": "locked"},
            )
        ],
        locked=True,
        lock_reason="staged storyboard",
        locked_by="JT",
        locked_at="2026-06-13T00:00:00Z",
    )
    path = scene_path(PROJECT, "ep_001", BATCH_ID)
    save_scene(scene, path)
    return path


def test_rederive_cli_stage_c_dry_run_is_read_only_with_inert_lock(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """REC-231 Phase 7: a dry-run rederive stays fully read-only (REC-100) and the
    lock is inert — a locked scene is no longer reported as a SKIP (it is a would-write
    candidate like any other), so the summary is ``written=4 skipped=0`` with NOTHING
    persisted to disk (no scene body, locked target byte-untouched)."""
    _write_plan()
    locked_path = _save_locked_scene()
    locked_before = locked_path.read_bytes()

    monkeypatch.setattr(
        generate.sys,
        "argv",
        [
            "generate.py",
            "rederive",
            "--project",
            PROJECT,
            "--episode",
            "1",
            "--skip-camera-test",
            "--skip-plan",
            "--skip-extract",
            "--dry-run",
        ],
    )

    code = generate.main()
    captured = capsys.readouterr()

    assert code == 0
    assert "COST: free CLI lane (Opus OAuth)" in captured.out
    assert (
        "rederive project=fixture episode=ep_001 written=4 skipped=0"
        in captured.out
    )
    assert "SKIP BATCH_004" not in captured.out
    # Dry-run persists nothing: the locked target is byte-untouched and no sibling
    # scene body was written to disk.
    assert locked_path.read_bytes() == locked_before
    assert not scene_path(PROJECT, "ep_001", "BATCH_001").exists()


# ── REC-231 Phase 6 helpers ───────────────────────────────────────────────────────
def _board(artifact: str, status: str = "proposed") -> dict:
    return {
        "status": status,
        "artifact": artifact,
        "source_sha256": "d" * 64,
        "approved_by": None,
        "updated_at": "2026-06-22T00:00:00Z",
        "fingerprint_version": 1,
    }


def _r2v_beat(beat_id: str, *, board: dict | None = None) -> Beat:
    return Beat(
        beat_id=beat_id,
        board=board,
        beat_metadata={"modality": "r2v_multi", "grouping": dict(_GROUPING)},
    )


def _versioned_scene(beat_id: str, artifact: str, *, scene_id: str = BATCH) -> Scene:
    return Scene(
        scene_id=scene_id,
        beats=[_r2v_beat(beat_id, board=_board(artifact))],
        scene_metadata={"grouping": dict(_GROUPING)},
    )


def _entry(manifest: dict, version: int) -> dict:
    return next(e for e in manifest["versions"] if e["version"] == version)


def _rederive_shot(batch_index: int, shot_index: int) -> CanonicalShot:
    ordinal = (batch_index - 1) * 3 + shot_index
    return CanonicalShot(
        shot_id=f"EP001_SH{ordinal:02d}",
        scene_index=batch_index,
        sequence_id=None,
        pipeline="video",
        previs_model=None,
        video_model="seeddance-2.0",
        location_id=f"LOC_{batch_index}",
        characters=[],
        shot_type="MS",
        duration_s=2.0,
        is_env_only=False,
        has_dialogue=False,
        aspect_ratio="9:16",
        quality=None,
        cinematography=None,
        raw={},
    )


def _rederive_plan() -> CanonicalPlan:
    shots = [_rederive_shot(b, s) for b in range(1, 5) for s in range(1, 4)]
    return CanonicalPlan(
        episode_id="ep_001",
        project=PROJECT,
        shots=shots,
        source_path=Path("fixture_plan.json"),
        raw={"episode_id": "ep_001", "project": PROJECT,
             "shots": [{"shot_id": shot.shot_id} for shot in shots]},
    )


def _trap_run_scene():
    async def _run_scene(scene, **kwargs):  # noqa: ANN001, ANN003
        raise AssertionError("run_scene must not be called in derive_only mode")
    return _run_scene


# ── 1. operator loop: propose → conform → revert (live reader observes the pointer) ─
def test_operator_loop_propose_conform_revert() -> None:
    """LIVE end-to-end: propose (append candidate, pointer unmoved) → conform → revert,
    with the live board read-model (get_board) observing every pointer move and prior
    work byte-intact at every step."""
    flat = scene_path(PROJECT, EP, BATCH)
    save_scene(_versioned_scene("V1", _ART_V1), flat)
    flat_before = flat.read_bytes()

    store = SceneVersionStore(PROJECT, EP)
    # PROPOSE — append a v2 candidate via the real rederive write site (the appender).
    store.write_scene_candidate(BATCH, _versioned_scene("V2", _ART_V2))
    v2_path = scene_version_path(PROJECT, EP, BATCH, 2)
    v2_before = v2_path.read_bytes()

    # pointer unmoved → the live reader still shows the v1 board graph
    assert load_manifest(PROJECT, EP, BATCH)["active_version"] == 1
    assert readmodel.get_board(SELECTOR, PROJECT).board_artifact == _ART_V1

    # CONFORM → the live reader observes v2; v1 body byte-intact
    store.conform(BATCH, 2)
    assert readmodel.get_board(SELECTOR, PROJECT).board_artifact == _ART_V2
    assert flat.read_bytes() == flat_before

    # REVERT → the live reader observes v1 again, byte-intact; candidate untouched
    store.revert(BATCH, 1)
    assert readmodel.get_board(SELECTOR, PROJECT).board_artifact == _ART_V1
    assert flat.read_bytes() == flat_before
    assert v2_path.read_bytes() == v2_before


# ── 2. identity-halt writes nothing (non-tautological) ────────────────────────────
def test_identity_halt_writes_nothing() -> None:
    """The store HALTS on a renumber (requested BATCH_001 vs generated BATCH_999) and
    writes NOTHING — proving the requested-vs-generated comparison is real."""
    flat = scene_path(PROJECT, EP, BATCH)
    save_scene(_versioned_scene("V1", _ART_V1), flat)
    flat_before = flat.read_bytes()

    store = SceneVersionStore(PROJECT, EP)
    renumbered = _versioned_scene("X", _ART_V2, scene_id="BATCH_999")
    with pytest.raises(SceneIdentityMismatchError):
        store.write_scene_candidate(BATCH, renumbered)  # requested BATCH_001 ≠ BATCH_999

    assert not scene_manifest_path(PROJECT, EP, BATCH).exists()
    assert not scene_version_path(PROJECT, EP, BATCH, 2).exists()
    assert flat.read_bytes() == flat_before
    assert not scene_manifest_path(PROJECT, EP, "BATCH_999").exists()
    assert not scene_path(PROJECT, EP, "BATCH_999").exists()


# ── 2b. identity-halt on the LIVE rederive path (non-tautological end-to-end) ──────
def test_identity_halt_live_rederive(monkeypatch: pytest.MonkeyPatch) -> None:
    """LIVE path: the real run_episode_batches derive_only write site HALTS when the
    grouping renumbers (generated scene_id ≠ requested), proving the runner threads the
    PRE-derivation requested id (group.scene_id), not scene.scene_id, into the store."""
    flat = scene_path(PROJECT, EP, BATCH)
    save_scene(Scene(scene_id=BATCH, beats=[Beat(beat_id="OPENING")]), flat)
    flat_before = flat.read_bytes()

    plan = _rederive_plan()
    runner = EpisodeRunner(
        project=PROJECT, plan=plan.raw, casting={}, episode=EP, concurrency=1
    )
    monkeypatch.setattr(runner, "run_scene", _trap_run_scene())

    real_scene_from_group = runner._scene_from_group

    def _renumbering(group):  # the scene-clusterer renumbers — a NEW identity
        scene = real_scene_from_group(group)
        scene.scene_id = "BATCH_999"
        return scene

    monkeypatch.setattr(runner, "_scene_from_group", _renumbering)

    with pytest.raises(SceneIdentityMismatchError):
        asyncio.run(
            runner.run_episode_batches(plan, derive_only=True, only_scene_ids={BATCH})
        )

    # the live writer halted BEFORE any write — no body, no manifest entry
    assert not scene_manifest_path(PROJECT, EP, BATCH).exists()
    assert not scene_version_path(PROJECT, EP, BATCH, 2).exists()
    assert flat.read_bytes() == flat_before
    assert not scene_path(PROJECT, EP, "BATCH_999").exists()


# ── 3. kind recorded on the candidate entry (default reshoot; invalid → ValueError) ─
def test_kind_recorded_on_candidate() -> None:
    """write_scene_candidate records the caller-supplied kind (default reshoot) and
    rejects an invalid kind with ValueError (auto-classification is DEFERRED)."""
    save_scene(_versioned_scene("V1", _ART_V1), scene_path(PROJECT, EP, BATCH))
    store = SceneVersionStore(PROJECT, EP)

    store.write_scene_candidate(BATCH, _versioned_scene("V2", _ART_V2), kind="pickup")
    assert _entry(load_manifest(PROJECT, EP, BATCH), 2)["kind"] == "pickup"

    store.write_scene_candidate(BATCH, _versioned_scene("V3", _ART_V2))
    assert _entry(load_manifest(PROJECT, EP, BATCH), 3)["kind"] == "reshoot"  # default

    with pytest.raises(ValueError):
        store.write_scene_candidate(BATCH, _versioned_scene("V4", _ART_V2), kind="bogus")


# ── 4. conform/revert CLI parser matrix (subprocess-level) ────────────────────────
def test_conform_revert_cli_parser_matrix(_projects_root: Path) -> None:
    """Subprocess-level: --conform/--revert move the pointer (exit 0) and the --batch
    requirement matrix enforces the mutual exclusions (non-zero parser exit), not a
    silent no-op."""
    root = _projects_root.parent
    save_scene(_versioned_scene("V1", _ART_V1), scene_path(PROJECT, EP, BATCH))
    store = SceneVersionStore(PROJECT, EP)
    store.write_scene_candidate(BATCH, _versioned_scene("V2", _ART_V2))
    assert load_manifest(PROJECT, EP, BATCH)["active_version"] == 1

    gen_py = Path(generate.__file__)
    env = {**os.environ, "RECOIL_PROJECTS_ROOT": str(root)}

    def _cli(*extra: str) -> subprocess.CompletedProcess:
        return subprocess.run(
            [sys.executable, str(gen_py), "rederive",
             "--project", PROJECT, "--episode", "1", *extra],
            capture_output=True, text=True, env=env, timeout=180,
        )

    # valid: conform → v2 (exit 0)
    res = _cli("--conform", "--batch", SELECTOR, "--to-version", "2")
    assert res.returncode == 0, res.stderr
    assert load_manifest(PROJECT, EP, BATCH)["active_version"] == 2

    # valid: revert → v1 (exit 0)
    res = _cli("--revert", "--batch", SELECTOR, "--to-version", "1")
    assert res.returncode == 0, res.stderr
    assert load_manifest(PROJECT, EP, BATCH)["active_version"] == 1

    # invalid combos → non-zero parser exit (SystemExit(2)), never a silent no-op
    assert _cli("--conform", "--to-version", "2").returncode != 0  # no --batch
    assert _cli(
        "--conform", "--from-script", "--batch", SELECTOR, "--to-version", "2"
    ).returncode != 0  # conform + from-script (mutually exclusive)
    assert _cli(
        "--conform", "--revert", "--batch", SELECTOR, "--to-version", "2"
    ).returncode != 0  # conform + revert (mutually exclusive)

    # the rejected invalid combos never moved the pointer
    assert load_manifest(PROJECT, EP, BATCH)["active_version"] == 1