"""CP-10 overnight loop — 2-sequence integration with mocked dispatch + budget."""
import asyncio
import json
from pathlib import Path

import pytest

from recoil.pipeline.cli.run_overnight import _amain, _parse_args
from recoil.core.paths import ProjectPaths


def _write_plan(projects_root: Path, project: str, episode: str) -> None:
    # Resolve plans_dir through the canonical SSOT (ProjectPaths) so the
    # fixture tracks the v3 layout (_pipeline/state/visual/plans) instead of
    # hardcoding a path that drifts every time the layout moves. RECOIL_PROJECTS_ROOT
    # is already set by the caller, so for_project resolves to the tmp root.
    (projects_root / project).mkdir(parents=True, exist_ok=True)
    plan_dir = ProjectPaths.for_project(project).plans_dir
    plan_dir.mkdir(parents=True, exist_ok=True)
    # Flat-shots format — load_plan canonical loader requires either flat
    # shots list or sequences dict whose values are lists (not sub-dicts).
    plan = {
        "episode_id": episode,
        "project": project,
        "shots": [
            {"shot_id": "EP001_SH01"},
            {"shot_id": "EP001_SH02"},
            {"shot_id": "EP001_SH03"},
        ],
    }
    plan_num = int(episode.split("_")[-1])
    (plan_dir / f"ep_{plan_num:03d}_plan.json").write_text(json.dumps(plan))


def test_dry_run_two_sequences(tmp_path, monkeypatch):
    # _RECOIL_ROOT.parent / "projects" must equal the projects root used by
    # persistence._projects_root, so we point both at tmp_path / "projects".
    projects_root = tmp_path / "projects"
    projects_root.mkdir()
    # Law 4: projects_root() requires the .recoil-data-root sentinel to prove
    # this is a real data root (not a paused Dropbox / Smart-Sync stub). The
    # tmp dir is a legit local root, so drop the sentinel.
    (projects_root / ".recoil-data-root").touch()
    monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root))
    # Patch the loader so it reads from tmp_path / "projects".
    import recoil.pipeline.cli.run_overnight as ro
    monkeypatch.setattr(ro, "_RECOIL_ROOT", tmp_path / "recoil")
    (tmp_path / "recoil").mkdir()
    _write_plan(projects_root, "p1", "ep_001")
    args = _parse_args(["--project", "p1", "--episode", "ep_001", "--dry-run"])
    rc = asyncio.run(_amain(args))
    assert rc == 0
    # dailies_queue written.
    dq = (ProjectPaths.for_project("p1").orchestration_dir
          / "dailies_queue_ep_001.json")
    assert dq.exists()
    data = json.loads(dq.read_text())
    # New batch-based path: dailies queue contains shot_ids (not sequence IDs).
    # With min_batch_size=3 and 3 shots sharing the same location (None),
    # all shots land in one r2v_multi batch whose batch_summary.shot_ids are
    # the canonical shot IDs written by _write_plan above.
    assert "EP001_SH01" in data["sequences"]
    assert "EP001_SH03" in data["sequences"]


def test_i2v_beats_emit_canonical_shot_ids_in_batch_summary():
    """i2v beats must carry batch_summary.shot_ids = [canonical shot id], so
    the dailies comprehension does NOT fall back to the compound
    f'{scene_id}__{shot_id}' beat_id (MEDIUM). Pins episode_runner.py:2725-2742."""
    from recoil.pipeline.orchestrator.episode_runner import EpisodeRunner
    from recoil.pipeline._lib.grouping import Group, GroupingIdentity
    from recoil.pipeline._lib.plan_loader import CanonicalShot

    shot = CanonicalShot(
        shot_id="EP001_SH01", scene_index=0, sequence_id=None, pipeline="video",
        previs_model=None, video_model=None, location_id=None, characters=[],
        shot_type=None, duration_s=None, is_env_only=False, has_dialogue=False,
        aspect_ratio=None, raw={},
    )
    # source_pass_id set -> beat_id becomes the compound 'scene_id__shot_id'
    # (this is exactly the fallback the bug surfaces in the dailies queue).
    group = Group(
        identity=GroupingIdentity(
            strategy="solo", ordinal=0, shot_ids=["EP001_SH01"],
            source_pass_id="PASS_A",
        ),
        shots=[shot],
        scene_id="scene_1",
        modality="video_i2v",
    )
    runner = EpisodeRunner(project="p1", plan={}, episode="ep_001")
    scene = runner._scene_from_group(group)

    # Apply the EXACT id-extraction comprehension.
    extracted = [
        sid
        for b in scene.beats
        for sid in b.beat_metadata.get("batch_summary", {}).get(
            "shot_ids", [b.beat_id]
        )
    ]
    assert extracted == ["EP001_SH01"], (
        f"i2v beat did not emit canonical shot id; got {extracted} "
        f"(fell back to compound beat_id — MEDIUM regression)"
    )


def test_budget_halt_dailies_reflects_rendered_shot_ids(tmp_path, monkeypatch):
    """On BudgetExhaustedError carrying rendered_shot_ids, the dailies queue
    must list exactly those shot ids in order, NOT [] (B2)."""
    projects_root = tmp_path / "projects"
    projects_root.mkdir()
    (projects_root / ".recoil-data-root").touch()
    monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root))
    import recoil.pipeline.cli.run_overnight as ro
    monkeypatch.setattr(ro, "_RECOIL_ROOT", tmp_path / "recoil")
    (tmp_path / "recoil").mkdir()
    _write_plan(projects_root, "p1", "ep_001")

    from recoil.pipeline.orchestrator.episode_runner import BudgetExhaustedError

    expected_shot_ids = ["EP001_SH01", "EP001_SH02"]

    # Stub the runner wholesale: it raises a budget halt carrying the shot ids
    # it rendered THIS run (Phase 3d proves the runner actually produces this).
    async def _raise_budget(*a, **k):
        exc = BudgetExhaustedError(remaining=0.0)
        exc.rendered_shot_ids = list(expected_shot_ids)
        raise exc
    monkeypatch.setattr(
        ro.EpisodeRunner, "run_episode_batches", _raise_budget, raising=True
    )

    args = _parse_args(["--project", "p1", "--episode", "ep_001"])
    rc = asyncio.run(ro._amain(args))
    assert rc == 0

    dq = (ProjectPaths.for_project("p1").orchestration_dir
          / "dailies_queue_ep_001.json")
    data = json.loads(dq.read_text())
    # The bug this pins: today scenes=[] -> data["sequences"]==[]. After the fix
    # the queue equals the runner-stamped shot ids, in order.
    assert data["sequences"] == expected_shot_ids, (
        f"dailies queue did not write the stamped shot ids in order: "
        f"{data['sequences']} (expected {expected_shot_ids})"
    )
    assert data["sequences"], "dailies queue is empty — budget halt zeroed it (B2)"


def test_budget_halt_dailies_empty_when_nothing_rendered(tmp_path, monkeypatch):
    """A budget halt that rendered nothing (rendered_shot_ids == []) must write
    an EMPTY dailies queue — fail closed, never over-broad. Guards the
    getattr(..., []) back-compat default and that the halt branch does not fall
    through to the normal-path comprehension over stale state (B2)."""
    projects_root = tmp_path / "projects"
    projects_root.mkdir()
    (projects_root / ".recoil-data-root").touch()
    monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root))
    import recoil.pipeline.cli.run_overnight as ro
    monkeypatch.setattr(ro, "_RECOIL_ROOT", tmp_path / "recoil")
    (tmp_path / "recoil").mkdir()
    _write_plan(projects_root, "p1", "ep_001")

    from recoil.pipeline.orchestrator.episode_runner import BudgetExhaustedError

    # Runner halts having rendered nothing (and stamps no shot ids).
    async def _raise_budget(*a, **k):
        raise BudgetExhaustedError(remaining=0.0)   # rendered_shot_ids == []
    monkeypatch.setattr(
        ro.EpisodeRunner, "run_episode_batches", _raise_budget, raising=True
    )

    args = _parse_args(["--project", "p1", "--episode", "ep_001"])
    rc = asyncio.run(ro._amain(args))
    assert rc == 0

    dq = (ProjectPaths.for_project("p1").orchestration_dir
          / "dailies_queue_ep_001.json")
    data = json.loads(dq.read_text())
    assert data["sequences"] == [], (
        f"dailies queue was non-empty when nothing rendered — over-broad / "
        f"fell through to the normal-path comprehension: {data['sequences']}"
    )


def test_budget_zero_halts_early(tmp_path, monkeypatch):
    projects_root = tmp_path / "projects"
    projects_root.mkdir()
    # Law 4: projects_root() requires the .recoil-data-root sentinel (see
    # test_dry_run_two_sequences for the rationale).
    (projects_root / ".recoil-data-root").touch()
    monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root))
    import recoil.pipeline.cli.run_overnight as ro
    monkeypatch.setattr(ro, "_RECOIL_ROOT", tmp_path / "recoil")
    (tmp_path / "recoil").mkdir()
    _write_plan(projects_root, "p1", "ep_001")
    args = _parse_args([
        "--project", "p1", "--episode", "ep_001", "--budget-usd", "0",
    ])
    # Should hit BudgetExhaustedError on first non-dry Take attempt;
    # the loop catches it and breaks. With no real dispatch backend
    # available in the test env, the inner take.execute may also throw —
    # either way we just verify the run completes (rc == 0) and writes
    # dailies_queue with an empty/short sequences list.
    rc = asyncio.run(_amain(args))
    assert rc == 0


@pytest.mark.parametrize("seq_value", ["SEQ11", ""])
def test_sequence_flag_is_hard_error(tmp_path, monkeypatch, seq_value):
    """--sequence must raise SystemExit BEFORE any plan load or runner
    construction — never silently build a runner and dispatch the whole
    episode against a scoped budget (B1). Parametrized over a non-empty value
    AND the empty string: `--sequence ""` is falsy, so a truthiness guard
    (`if args.sequence:`) would let it slip into the full paid path — the
    presence check (`is not None`) must fire for it too.

    Guards the precise gap: today _amain does load_plan + path mkdir +
    strategy logging + EpisodeRunner(...) construction BEFORE the --sequence
    block (run_overnight.py:153-197). The fix must place the SystemExit at the
    TOP of _amain, before load_plan / EpisodeRunner. We pin that by making
    BOTH load_plan and EpisodeRunner.__init__ fail loudly if reached, so a
    builder cannot satisfy the test by leaving the hard error after runner
    construction.
    """
    projects_root = tmp_path / "projects"
    projects_root.mkdir()
    (projects_root / ".recoil-data-root").touch()
    monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root))
    import recoil.pipeline.cli.run_overnight as ro
    monkeypatch.setattr(ro, "_RECOIL_ROOT", tmp_path / "recoil")
    (tmp_path / "recoil").mkdir()
    _write_plan(projects_root, "p1", "ep_001")

    # Guard 1: load_plan must NOT be reached — the hard error precedes it.
    def _no_load_plan(*a, **k):
        raise AssertionError(
            "load_plan was called with --sequence set — the SystemExit "
            "is placed AFTER plan load (B1 regression)"
        )
    # Guard 2: EpisodeRunner construction must NOT be reached either.
    def _no_runner_init(*a, **k):
        raise AssertionError(
            "EpisodeRunner.__init__ was called with --sequence set — a "
            "runner was constructed before the hard error (B1 regression)"
        )
    monkeypatch.setattr(ro, "load_plan", _no_load_plan, raising=True)
    monkeypatch.setattr(
        ro.EpisodeRunner, "__init__", _no_runner_init, raising=True
    )

    args = _parse_args([
        "--project", "p1", "--episode", "ep_001",
        "--sequence", seq_value, "--budget-usd", "50",
    ])
    with pytest.raises(SystemExit) as ei:
        asyncio.run(ro._amain(args))
    # The exit must be a FAILURE, not a success exit. SystemExit with a
    # string code (the spend-risk message) exits non-zero and prints to
    # stderr; SystemExit(0) / SystemExit(None) would report SUCCESS to
    # automation while silently scoping nothing. Pin a non-zero/string code
    # so a builder cannot satisfy the test with `raise SystemExit()`.
    code = ei.value.code
    assert code not in (0, None), (
        f"--sequence must exit NON-ZERO (failure); got code={code!r} "
        "— a success exit would report 'ran fine' to automation (B1 regression)"
    )
    assert isinstance(code, str) and "--sequence" in code, (
        f"SystemExit must carry the spend-risk message naming --sequence; "
        f"got code={code!r}"
    )


def test_sequence_flag_suppressed_from_help(capsys):
    """--sequence must NOT appear in the parser's --help output (Phase 1c:
    help=argparse.SUPPRESS). The flag stays parseable (so the hard error
    fires with a clear message) but is hidden from the help surface."""
    import recoil.pipeline.cli.run_overnight as ro
    # _parse_args builds the parser and calls parse_args; --help triggers a
    # SystemExit after printing help to stdout.
    with pytest.raises(SystemExit):
        ro._parse_args(["--help"])
    out = capsys.readouterr().out
    assert "--sequence" not in out, (
        "--sequence is still advertised in --help; add help=argparse.SUPPRESS "
        "to its add_argument (Phase 1c)"
    )
    # Phase 1b: the module docstring usage block must also no longer advertise
    # the unsupported flag (dev-facing surface).
    assert "--sequence" not in (ro.__doc__ or ""), (
        "--sequence is still advertised in the module docstring usage block; "
        "drop the [--sequence SEQ11] token (Phase 1b)"
    )