"""Lease-safe actuation: send commands only while the lease is confirmed; stop on your own
clock when it isn't. Python 3.8+, standard library only (fits a ROS 2 node or any robot loop).
Set CASCADE_API_KEY (and CASCADE_API for a local server).

Cascade is a coordination layer, NOT a safety-rated system. Cloud round trips are ~150 ms
at the median and worse in the tail: use it for task / zone level allocation (who may work
in aisle 7 for the next minute), never for motion control, collision avoidance or e-stop.

Rule: renew every ttl/3; after a successful renewal the lease is good until
(time the renewal was SENT) + ttl - margin on time.monotonic(). When that passes without a
newer renewal, on_stop(reason) runs once and every further command is refused.
"""
import json, os, threading, time, urllib.error, urllib.request


def cascade(path, body, timeout=15.0):
    api = os.environ.get("CASCADE_API", "https://api.cascadedynamics.ai/v1")
    req = urllib.request.Request(api + path, data=json.dumps(body).encode(), method="POST", headers={
        "Authorization": "Bearer " + os.environ["CASCADE_API_KEY"],
        "Content-Type": "application/json", "User-Agent": "cascade-recipes/2.0"})
    with urllib.request.urlopen(req, timeout=timeout + body.get("wait_ms", 0) / 1000) as r:
        return json.loads(r.read())


class LeaseLost(Exception):
    pass


class Lease:
    def __init__(self, cell, agent, r, sent_at, ttl_ms, margin_ms, on_stop):
        self.cell, self.agent, self.ttl, self.margin, self.on_stop = cell, agent, ttl_ms / 1000, margin_ms / 1000, on_stop
        self.fence, self.receipt = r["fence"], r["entry_hash"]
        self.deadline = sent_at + self.ttl - self.margin
        self.stopped = None
        self.renewals = 0
        self._lock = threading.RLock()
        self._halt = threading.Event()
        self._renewer = threading.Thread(target=self._renew_loop, daemon=True)
        self._watcher = threading.Thread(target=self._watch_loop, daemon=True)
        self._renewer.start()
        self._watcher.start()

    def _watch_loop(self):  # reads only the local clock: this is what stops the machine
        tick = max(0.01, min(0.05, self.margin / 4))
        while not self._halt.wait(tick):
            if time.monotonic() >= self.deadline:
                self._stop("lease not confirmed before its deadline")

    def _renew_loop(self):
        while not self._halt.wait(self.ttl / 3):
            sent_at = time.monotonic()
            try:
                r = cascade("/claim", {"cell": self.cell, "agent": self.agent, "ttl_ms": int(self.ttl * 1000)}, timeout=self.ttl / 3)
            except Exception:
                continue  # no answer: the watchdog decides
            with self._lock:
                if self.stopped:
                    return
                if not r.get("ok"):
                    return self._stop(f"lost to {r.get('behind')}")
                if r["fence"] != self.fence:
                    return self._stop(f"hold lapsed and restarted (fence {self.fence} -> {r['fence']})")
                self.deadline = max(self.deadline, sent_at + self.ttl - self.margin)
                self.renewals += 1

    def _stop(self, reason):
        with self._lock:
            if self.stopped:
                return
            self.stopped = reason
        self._halt.set()
        try:
            self.on_stop(reason)
        except Exception:
            pass

    def valid(self):
        return not self.stopped and time.monotonic() < self.deadline

    def command(self, send):
        """send(fence) runs only if the lease is confirmed right now."""
        if not self.valid():
            self._stop("lease not confirmed before its deadline")
            raise LeaseLost(self.stopped)
        return send(self.fence)

    def release(self, outcome=None):
        """Only once the machine is out of the zone or parked safely."""
        with self._lock:
            was = self.stopped
            self.stopped = self.stopped or "released"
        self._halt.set()
        if threading.current_thread() is not self._renewer:
            self._renewer.join()  # a renewal on the wire must not land after the release
        try:
            r = cascade("/release", {"cell": self.cell, "agent": self.agent, **({"outcome": outcome} if outcome else {})})
        except Exception as e:
            r = {"released": False, "error": str(e)}
        return {**r, "stopped_early": was}


def acquire_lease(cell, agent, on_stop, ttl_ms=6000, margin_ms=None, wait_ms=0):
    """Lease, or {"ok": False, "behind"} when another agent holds the cell."""
    margin_ms = margin_ms if margin_ms is not None else ttl_ms // 4
    if not 0 < margin_ms < ttl_ms * 2 / 3:
        raise ValueError("margin_ms must be > 0 and < 2/3 of ttl_ms")
    sent_at = time.monotonic()
    r = cascade("/claim", {"cell": cell, "agent": agent, "ttl_ms": ttl_ms, **({"wait_ms": wait_ms} if wait_ms else {})})
    if not r.get("ok"):
        return {"ok": False, "behind": r.get("behind")}
    return Lease(cell, agent, r, sent_at + r.get("waited_ms", 0) / 1000, ttl_ms, margin_ms, on_stop)