"""Cross-process task lock.

An in-process `threading.Lock` only serialises one interpreter. Two Uvicorn workers, or a
CLI run alongside a running server, would otherwise advance the same task at once: two
real provider calls for one stage, and two writers racing on state.json.

ponytail: a pid-stamped lock file, not a lock server. It covers every process on this
host, which is the whole deployment the design describes. A shared runtime directory over
NFS would need real fcntl locking instead.
"""

from __future__ import annotations

import json
import os
import time
from contextlib import contextmanager
from pathlib import Path

from .infra.process import is_running


class TaskBusy(RuntimeError):
    """Another process holds this task."""


class FileLock:
    def __init__(self, path: str | Path, stale_after_s: float = 3600.0) -> None:
        self.path = Path(path)
        self.stale_after_s = stale_after_s
        self._acquired = False

    def _holder(self) -> dict | None:
        try:
            return json.loads(self.path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return None

    def _clear_if_dead(self) -> bool:
        """Remove a lock whose owner is gone, or which outlived the stale window."""
        holder = self._holder()
        if holder is None:
            # Unreadable but present: only reclaim it once it is older than the window,
            # so a half-written lock file is not stolen from a live writer.
            try:
                age = time.time() - self.path.stat().st_mtime
            except OSError:
                return True
            if age < 5:
                return False
            self.path.unlink(missing_ok=True)
            return True
        pid = holder.get("pid")
        age = time.time() - float(holder.get("at", 0))
        if (isinstance(pid, int) and not is_running(pid)) or age > self.stale_after_s:
            self.path.unlink(missing_ok=True)
            return True
        return False

    def acquire(self, timeout_s: float = 0.0) -> bool:
        deadline = time.monotonic() + timeout_s
        payload = json.dumps({"pid": os.getpid(), "at": time.time()}).encode("utf-8")
        while True:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            try:
                handle = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            except FileExistsError:
                if self._clear_if_dead():
                    continue
                if time.monotonic() >= deadline:
                    return False
                time.sleep(0.05)
                continue
            try:
                os.write(handle, payload)
            finally:
                os.close(handle)
            self._acquired = True
            return True

    def release(self) -> None:
        if not self._acquired:
            return
        holder = self._holder()
        if holder is None or holder.get("pid") == os.getpid():
            self.path.unlink(missing_ok=True)
        self._acquired = False

    def holder_pid(self) -> int | None:
        holder = self._holder()
        pid = holder.get("pid") if holder else None
        return pid if isinstance(pid, int) else None


@contextmanager
def task_lock(directory: str | Path, timeout_s: float = 0.0):
    """Hold the lock for one task directory, or raise TaskBusy naming the holder."""
    lock = FileLock(Path(directory) / "task.lock")
    if not lock.acquire(timeout_s):
        raise TaskBusy(f"another process (pid {lock.holder_pid()}) is advancing this task")
    try:
        yield lock
    finally:
        lock.release()