"""Late settlement requires a real executing transition (R4 item 4).

An admitted status also describes pending, never-dispatched work, so the
async-completion seam must demand the executing marker and refuse
contradictory pending/executing state before persisting a late result.
Settling a stopped run twice is rejected; ownership stays enforced.
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, cast

import pytest

from tests.scheduling.controlled import (
    DictDeployments,
    ScriptedDispatcher,
    fixture_environment,
)
from tests.scheduling.controlled import ScriptedDispatcher as SD
from wf_artifacts.runs.store import FileRunStore
from wf_scheduling.calendar import OneShotSource
from wf_scheduling.models import Schedule
from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError
from wf_scheduling.poll import BlockedSchedule, Scheduler
from wf_scheduling.prepare import SchedulePreparer
from wf_scheduling.store import FileScheduleStore


def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0) -> datetime:
    return datetime(y, mo, d, h, mi, tzinfo=UTC)


def _sched_model(sid: str, **kw: Any) -> Schedule:
    now = ts(2026, 9, 8, 12, 0)
    base: dict[str, Any] = {
        "id": sid,
        "deployment_id": "dep-1",
        "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"},
        "input_bindings": [],
        "created_at": now.isoformat(),
        "updated_at": now.isoformat(),
    }
    base.update(kw)
    return Schedule.model_validate(base)


def _harness(
    root: Path,
    ownership: SchedulerOwnership,
    *,
    script: dict | None = None,
) -> tuple[Scheduler, FileScheduleStore, FileRunStore]:
    sched_store = FileScheduleStore(root / "sched")
    run_store = FileRunStore(root / "runs")
    sched = Scheduler(
        schedule_store=sched_store,
        run_store=run_store,
        sources={},
        capacity=4,
        preparer=SchedulePreparer(
            DictDeployments({"dep-1": {"rev": 1, "required": []}}),
            fixture_environment,
        ),
        dispatcher=ScriptedDispatcher(script),
        ownership=ownership,
    )
    return sched, sched_store, run_store


def _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None:
    store.create_schedule(_sched_model("a"))
    store.save_consumed("a", intended - timedelta(hours=1))
    sched.sources["a"] = OneShotSource(intended)


def test_pending_undispatched_run_rejects_settlement(tmp_path: Path) -> None:
    from tests.artifacts.test_run_store import artifact as _artifact
    from tests.artifacts.test_run_store import deployment as _deployment
    from wf_api.run_lifecycle import materialize_admitted_view, persist_admission
    from wf_artifacts import PinnedRunEnvironment
    from wf_core import RunState, RunStatus

    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
    try:
        sched, _, runs = _harness(tmp_path, ownership, script={"*": "hang"})
        admission = persist_admission(
            store=runs,
            run_id=runs.allocate_run_id(),
            environment=PinnedRunEnvironment(
                deployment=_deployment(),
                root_artifact=_artifact(),
                child_artifacts=[],
            ),
            resolved_input={},
            max_steps=None,
            scheduled_at=ts(2026, 9, 8, 12, 0),
            schedule_id="a",
            schedule_revision=1,
        )
        materialize_admitted_view(store=runs, admission=admission)
        runs.mark_pending_dispatch(admission.id)
        state = RunState(
            workflow_name="sched",
            status=RunStatus.COMPLETED,
            workflow_input={},
            state={},
        )
        with pytest.raises(BlockedSchedule):
            sched.record_stopped_execution(admission.id, state, ts(2026, 9, 8, 12, 0))
        assert runs.get_run(admission.id).status.value == "admitted"
        assert runs.is_pending_dispatch(admission.id)
        assert not runs.is_executing(admission.id)
        with pytest.raises(KeyError):
            runs.get_latest_checkpoint(admission.id)
    finally:
        ownership.release()


def test_missing_executing_marker_rejects_settlement(tmp_path: Path) -> None:
    from tests.artifacts.test_run_store import artifact as _artifact
    from tests.artifacts.test_run_store import deployment as _deployment
    from wf_api.run_lifecycle import materialize_admitted_view, persist_admission
    from wf_artifacts import PinnedRunEnvironment
    from wf_core import RunState, RunStatus

    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
    try:
        sched, _, runs = _harness(tmp_path, ownership, script={"*": "hang"})
        admission = persist_admission(
            store=runs,
            run_id=runs.allocate_run_id(),
            environment=PinnedRunEnvironment(
                deployment=_deployment(),
                root_artifact=_artifact(),
                child_artifacts=[],
            ),
            resolved_input={},
            max_steps=None,
            scheduled_at=ts(2026, 9, 8, 12, 0),
            schedule_id="a",
            schedule_revision=1,
        )
        materialize_admitted_view(store=runs, admission=admission)
        assert not runs.is_pending_dispatch(admission.id)
        state = RunState(
            workflow_name="sched",
            status=RunStatus.COMPLETED,
            workflow_input={},
            state={},
        )
        with pytest.raises(BlockedSchedule):
            sched.record_stopped_execution(admission.id, state, ts(2026, 9, 8, 12, 0))
        assert runs.get_run(admission.id).status.value == "admitted"
        with pytest.raises(KeyError):
            runs.get_latest_checkpoint(admission.id)
    finally:
        ownership.release()


def test_contradictory_pending_and_executing_rejects_settlement(
    tmp_path: Path,
) -> None:
    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
    try:
        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
        intended = ts(2026, 9, 8, 12, 0)
        _due(sched, store, intended)
        sched.poll(intended)
        run_id = runs.list_runs()[0].id
        assert runs.is_executing(run_id)
        # Crash shape between mark_executing and pending-clear: both markers.
        runs.mark_pending_dispatch(run_id)
        dispatcher = cast(SD, sched.dispatcher)
        admission = runs.get_admission(run_id)
        with pytest.raises(BlockedSchedule):
            sched.record_stopped_execution(
                run_id, dispatcher.finish(admission, "complete"), intended
            )
        assert runs.get_run(run_id).status.value == "admitted"
        assert runs.is_executing(run_id)
        assert runs.is_pending_dispatch(run_id)
        with pytest.raises(KeyError):
            runs.get_latest_checkpoint(run_id)
    finally:
        ownership.release()


def test_genuine_execution_settles_successfully(tmp_path: Path) -> None:
    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
    try:
        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
        intended = ts(2026, 9, 8, 12, 0)
        _due(sched, store, intended)
        sched.poll(intended)
        run_id = runs.list_runs()[0].id
        dispatcher = cast(SD, sched.dispatcher)
        admission = runs.get_admission(run_id)
        sched.record_stopped_execution(
            run_id, dispatcher.finish(admission, "complete"), intended
        )
        assert runs.get_run(run_id).status.value == "completed"
        assert not runs.is_executing(run_id)
    finally:
        ownership.release()


def test_duplicate_settlement_cannot_overwrite_stopped_result(
    tmp_path: Path,
) -> None:
    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
    try:
        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
        intended = ts(2026, 9, 8, 12, 0)
        _due(sched, store, intended)
        sched.poll(intended)
        run_id = runs.list_runs()[0].id
        dispatcher = cast(SD, sched.dispatcher)
        admission = runs.get_admission(run_id)
        sched.record_stopped_execution(
            run_id, dispatcher.finish(admission, "complete"), intended
        )
        with pytest.raises(BlockedSchedule):
            sched.record_stopped_execution(
                run_id, dispatcher.finish(admission, "interrupt"), intended
            )
        assert runs.get_run(run_id).status.value == "completed"
        assert len(runs.list_checkpoints(run_id)) == 1
    finally:
        ownership.release()


def test_settle_without_ownership_rejected(tmp_path: Path) -> None:
    ownership = SchedulerOwnership(tmp_path, owner="test").acquire()
    try:
        sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"})
        intended = ts(2026, 9, 8, 12, 0)
        _due(sched, store, intended)
        sched.poll(intended)
        run_id = runs.list_runs()[0].id
        dispatcher = cast(SD, sched.dispatcher)
        admission = runs.get_admission(run_id)
        ownership.release()
        with pytest.raises(SecondOwnerError):
            sched.record_stopped_execution(
                run_id, dispatcher.finish(admission, "complete"), intended
            )
        assert runs.get_run(run_id).status.value == "admitted"
    finally:
        ownership.release()