"""A slow skill refresh must not keep all later tool workers waiting forever."""

from __future__ import annotations

import concurrent.futures
import errno
import threading
from pathlib import Path

import pytest

from deerflow.sandbox import skill_view


def test_process_lock_wait_times_out_and_recovers(tmp_path, monkeypatch):
    monkeypatch.setattr(skill_view, "_REBUILD_WAIT_SECONDS", 0.03, raising=False)
    lock = threading.Lock()
    monkeypatch.setattr(skill_view, "_PROCESS_LOCK", lock)
    lock.acquire()
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
            future = pool.submit(lambda: skill_view._view_lock(tmp_path).__enter__())
            try:
                with pytest.raises(TimeoutError, match="skill.*lock"):
                    future.result(timeout=1)
            finally:
                lock.release()
    finally:
        if lock.locked():
            lock.release()
    with skill_view._view_lock(tmp_path):
        assert lock.locked()
    assert not lock.locked()


def test_warm_follower_timeout_does_not_replace_running_leader(monkeypatch):
    monkeypatch.setattr(skill_view, "_VIEW_WAIT_SECONDS", 0.03, raising=False)
    entered, release = threading.Event(), threading.Event()
    entry = object()
    key = ("source", "target", "scope")
    flights = {}
    monkeypatch.setattr(skill_view, "_WARM_VALIDATION_FLIGHTS", flights)

    def validate(*args):
        entered.set()
        assert release.wait(2)
        return True

    monkeypatch.setattr(skill_view, "_entry_matches", validate)
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
        leader = pool.submit(skill_view._entry_matches_singleflight, key, entry, Path("."), "config")
        assert entered.wait(1)
        follower = pool.submit(skill_view._entry_matches_singleflight, key, entry, Path("."), "config")
        try:
            with pytest.raises(TimeoutError, match="skill.*validation"):
                follower.result(timeout=1)
            assert len(flights) == 1
            assert not leader.done()
        finally:
            release.set()
        assert leader.result(timeout=1) is True
    assert not flights
    assert skill_view._entry_matches_singleflight(key, entry, Path("."), "config") is True


def test_file_lock_contention_is_bounded_and_process_lock_released(tmp_path, monkeypatch):
    monkeypatch.setattr(skill_view, "_REBUILD_WAIT_SECONDS", 0.03, raising=False)

    def busy(*args):
        raise OSError(errno.EAGAIN, "lock held by another worker")

    if skill_view.os.name == "nt":
        import msvcrt

        monkeypatch.setattr(msvcrt, "locking", busy)
    else:
        import fcntl

        monkeypatch.setattr(fcntl, "flock", busy)
    with pytest.raises(TimeoutError, match="skill.*lock"):
        with skill_view._view_lock(tmp_path):
            pytest.fail("must not enter the protected view")
    assert not skill_view._process_lock_held()


def test_same_thread_reenters_view_lock_without_waiting(tmp_path, monkeypatch):
    monkeypatch.setattr(skill_view, "_REBUILD_WAIT_SECONDS", 0.03, raising=False)
    with skill_view._view_lock(tmp_path):
        with skill_view._view_lock(tmp_path):
            assert getattr(skill_view._VIEW_LOCK_DEPTH, "depth", 0) == 2
    assert getattr(skill_view._VIEW_LOCK_DEPTH, "depth", 0) == 0


def test_file_lock_non_contention_error_is_not_retried(tmp_path, monkeypatch):
    def broken(*args):
        raise OSError(errno.EIO, "disk failure")

    if skill_view.os.name == "nt":
        import msvcrt

        monkeypatch.setattr(msvcrt, "locking", broken)
    else:
        import fcntl

        monkeypatch.setattr(fcntl, "flock", broken)
    with pytest.raises(OSError, match="disk failure"):
        with skill_view._view_lock(tmp_path):
            pytest.fail("must not enter the protected view")
    assert not skill_view._process_lock_held()