"""Lifecycle contracts only; no model, scheduler or simulated inference claim."""
from contextlib import contextmanager
from contextvars import ContextVar
from pathlib import Path
import threading
from types import SimpleNamespace

import pytest
from gradio.context import LocalContext
import workflow_artifacts as artifacts
import workflow_execution as execution


def arguments(path):
    return (["A", "A"], ["segment1", "segment2"], [90, 90], [0, 5], [90, 85],
            512, 512, [None, None], 42, ["array1", "array2"], ["anchor1", "anchor2"], str(path))


@contextmanager
def queued_context():
    event = SimpleNamespace(alive=True)
    queue = SimpleNamespace(event_ids_to_events={"event": event})
    values = {"request": object(), "event_id": "event",
              "blocks": SimpleNamespace(_queue=queue)}
    tokens = [(getattr(LocalContext, k), getattr(LocalContext, k).set(v)) for k, v in values.items()]
    try:
        yield values, event
    finally:
        for var, token in reversed(tokens):
            var.reset(token)


def test_context_and_all_windows_preserved_in_one_call(tmp_path):
    calls = []
    marker = ContextVar("marker", default="missing")
    marker.set("inherited")
    with queued_context() as (values, event):
        def render(*args):
            assert marker.get() == "inherited"
            for key, value in values.items():
                assert getattr(LocalContext, key).get(None) is value
            calls.append(args)
            return "unaltered-result"
        bridge = execution.GPUBridge(render, lambda *args: 1)
        args = arguments(tmp_path / "progress.json")
        assert bridge(*args, None) == "unaltered-result"
        assert calls == [args]
        bridge.ensure_idle()


def test_error_propagation_and_no_false_success(tmp_path):
    error = RuntimeError("original scheduler/model failure")
    def fail(*args):
        raise error
    progress = []
    bridge = execution.GPUBridge(fail, lambda *args: 1)
    with pytest.raises(RuntimeError) as raised:
        bridge(*arguments(tmp_path / "p.json"), lambda *a, **kw: progress.append(kw))
    assert raised.value is error
    assert not any(p.get("desc") == "Viggle — done" for p in progress)
    bridge.ensure_idle()


def test_timeout_blocks_new_jobs_until_real_worker_returns(tmp_path, monkeypatch):
    release = threading.Event()
    def held(*args):
        assert release.wait(2), "test worker release missing"
        return "late result"
    bridge = execution.GPUBridge(held, lambda *a: .03, grace_seconds=0, poll_seconds=.005)
    monkeypatch.setattr(execution, "_bridge", bridge)
    try:
        with pytest.raises(TimeoutError, match="waiting_worker"):
            artifacts.cached("timeout_test", [], lambda: ((bridge(*arguments(tmp_path / "p.json"), None),), []))
        with pytest.raises(execution.ExecutionBusy):
            artifacts.cached("other_model_stage", [], lambda: pytest.fail("model overlap"))
        assert not (artifacts.STATE / "cache" / (artifacts.digest(["timeout_test", []]) + ".json")).exists()
    finally:
        done = bridge.pending
        release.set()
        if done:
            assert done.wait(2)
    assert artifacts.cached("timeout_test", [], lambda: (("retry after worker exit",), [])) == ("retry after worker exit",)


def test_cancelled_result_never_commits_after_cpu_stage(tmp_path):
    with queued_context() as (values, event):
        def cancelled_compute():
            values["blocks"]._queue.event_ids_to_events.clear()
            return ("cancelled-result",), []
        with pytest.raises(execution.ExecutionCancelled):
            artifacts.cached("cancelled_cpu_test", [], cancelled_compute)
    assert artifacts.cached("cancelled_cpu_test", [], lambda: (("retry",), [])) == ("retry",)


def test_waiting_progress_does_not_claim_denoising(tmp_path):
    release = threading.Event()
    progress = []
    def worker(*args):
        assert release.wait(2)
        return "ok"
    def update(*args, **kw):
        progress.append(kw["desc"])
        release.set()
    bridge = execution.GPUBridge(worker, lambda *a: 1)
    try:
        assert bridge(*arguments(tmp_path / "absent.json"), update) == "ok"
    finally:
        release.set()
    assert "model entry not confirmed" in progress[0]
    assert not any("denoise" in p for p in progress)


def test_native_stage_lock_rejects_instead_of_waiting():
    held, release = threading.Event(), threading.Event()
    def occupy():
        with artifacts.EXECUTION_LOCK:
            held.set()
            release.wait(2)
    thread = threading.Thread(target=occupy)
    thread.start()
    try:
        assert held.wait(2)
        with pytest.raises(execution.ExecutionBusy, match="Another V22 stage"):
            artifacts.cached("lock_test", [], lambda: pytest.fail("must not run"))
    finally:
        release.set()
        thread.join(2)