"""Reserve all or nothing: every resource a job needs, in ONE decision (up to 16 cells).
Python 3.8+, standard library only. Set CASCADE_API_KEY (and CASCADE_API for a local server).

Claiming resources one at a time deadlocks (A holds dock/3 waiting for crane/1, B the
reverse). A multi-claim is granted whole or not at all, so nobody sits on half a set.
"""
import json, os, random, threading, time, urllib.error, urllib.request

MAX_CELLS, MAX_WAIT_MS = 16, 10000


def cascade(path, body):
    api = os.environ.get("CASCADE_API", "https://api.cascadedynamics.ai/v1")
    for attempt in range(6):
        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"})
        try:
            with urllib.request.urlopen(req, timeout=15 + body.get("wait_ms", 0) / 1000) as r:
                return json.loads(r.read())
        except urllib.error.HTTPError as e:
            if e.code in (429, 503) and attempt < 5:
                time.sleep(0.25 * 2 ** attempt)
                continue
            raise


def reserve_all(cells, agent, ttl_ms=30000, wait_ms=8000, timeout_ms=60000, on_hold=None):
    """{"ok": True, "cells", "fences", "receipts"} or {"ok": False, "cell", "behind"}.
    on_hold({"cell", "behind"}) may return False to stop waiting (e.g. re-route)."""
    cells = list(dict.fromkeys(cells))
    if not 1 <= len(cells) <= MAX_CELLS:
        raise ValueError(f"reserve_all needs 1-{MAX_CELLS} cells, got {len(cells)}")
    deadline = time.monotonic() + timeout_ms / 1000
    while True:
        left = int((deadline - time.monotonic()) * 1000)
        r = cascade("/claim", {"cells": cells, "agent": agent, "ttl_ms": ttl_ms,
                               "wait_ms": max(0, min(wait_ms, MAX_WAIT_MS, left))})
        if r.get("ok"):
            return {"ok": True, "cells": cells,
                    "fences": {g["cell"]: g["fence"] for g in r["grants"]},
                    "receipts": {g["cell"]: g["entry_hash"] for g in r["grants"]}}
        hold = {"cell": r.get("cell"), "behind": r.get("behind")}
        if (on_hold and on_hold(hold) is False) or time.monotonic() >= deadline:
            return {"ok": False, **hold}
        if not wait_ms:
            time.sleep(0.25 + random.random() * 0.25)


def release_all(cells, agent, outcome=None):
    out = []
    for cell in dict.fromkeys(cells):
        try:
            out.append(cascade("/release", {"cell": cell, "agent": agent, **({"outcome": outcome} if outcome else {})}))
        except Exception as e:
            out.append({"released": False, "error": str(e)})
    return out


def with_all(cells, agent, fn, ttl_ms=30000, outcome=None, **opts):
    """fn(lost_event, res) runs while the whole set is renewed every ttl/3. Returns
    {"ran": False, "cell", "behind"} if the set was not granted in time."""
    res = reserve_all(cells, agent, ttl_ms=ttl_ms, **opts)
    if not res["ok"]:
        return {"ran": False, "cell": res["cell"], "behind": res["behind"]}
    lost, halt = threading.Event(), threading.Event()

    def renew():
        while not halt.wait(ttl_ms / 3000):
            try:
                r = cascade("/claim", {"cells": res["cells"], "agent": agent, "ttl_ms": ttl_ms})
                if not r.get("ok") or any(g["fence"] != res["fences"][g["cell"]] for g in r["grants"]):
                    lost.set()
            except Exception:
                pass  # a blip; if it never recovers the TTL lapses
    t = threading.Thread(target=renew, daemon=True)
    t.start()
    try:
        return {"ran": True, "result": fn(lost, res), "fences": res["fences"], "receipts": res["receipts"]}
    finally:
        halt.set()
        t.join()
        release_all(res["cells"], agent, outcome)