"""ACP Python SDK - standard library only (Python 3.9+).

    from acp import ACP
    acp = ACP(api_key=os.environ["ACP_API_KEY"])
    run = acp.agents.run("researcher", input={"topic": "H100 pricing"}, wait=True)
    print(run["output"])
"""
from __future__ import annotations

import hashlib
import hmac
import json
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Callable, Dict, Iterator, List, Optional

__all__ = ["ACP", "ACPError", "verify_webhook"]
__version__ = "0.1.0"

TERMINAL = ("succeeded", "failed", "cancelled")


class ACPError(Exception):
    def __init__(self, status: int, code: str, message: str, details: Any = None):
        super().__init__(f"{status} {code}: {message}")
        self.status, self.code, self.message, self.details = status, code, message, details


def _q(params: Dict[str, Any]) -> str:
    items = {k: v for k, v in params.items() if v is not None}
    return ("?" + urllib.parse.urlencode(items)) if items else ""


def _p(s: str) -> str:
    return urllib.parse.quote(s, safe="")


class _Group:
    def __init__(self, client: "ACP"):
        self._c = client


class _Agents(_Group):
    def list(self, status: Optional[str] = None) -> List[dict]:
        return self._c.request("GET", "/agents" + _q({"status": status}))["agents"]

    def create(self, name: str, config: Optional[dict] = None, description: str = "", deploy: bool = False) -> dict:
        return self._c.request("POST", "/agents", {"name": name, "description": description, "config": config, "deploy": deploy})

    def get(self, ref: str) -> dict:
        return self._c.request("GET", f"/agents/{_p(ref)}")

    def update(self, ref: str, **patch: Any) -> dict:
        return self._c.request("PATCH", f"/agents/{_p(ref)}", patch)

    def deploy(self, ref: str) -> dict: return self._c.request("POST", f"/agents/{_p(ref)}/deploy")
    def pause(self, ref: str) -> dict: return self._c.request("POST", f"/agents/{_p(ref)}/pause")
    def resume(self, ref: str) -> dict: return self._c.request("POST", f"/agents/{_p(ref)}/resume")
    def archive(self, ref: str) -> dict: return self._c.request("POST", f"/agents/{_p(ref)}/archive")

    def run(self, ref: str, input: Optional[dict] = None, idempotency_key: Optional[str] = None, wait: bool = False, timeout: float = 300) -> dict:
        headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
        run = self._c.request("POST", f"/agents/{_p(ref)}/runs", {"input": input or {}}, headers)
        return self._c.wait_for_run(run["id"], timeout=timeout) if wait else run


class _Runs(_Group):
    def create(self, agent: Optional[str] = None, workflow: Optional[dict] = None, input: Optional[dict] = None,
               idempotency_key: Optional[str] = None, wait: bool = False, timeout: float = 300) -> dict:
        body = {"agent": agent, "workflow": workflow, "input": input or {}, "idempotencyKey": idempotency_key}
        run = self._c.request("POST", "/runs", {k: v for k, v in body.items() if v is not None})
        return self._c.wait_for_run(run["id"], timeout=timeout) if wait else run

    def list(self, status: Optional[str] = None, agent: Optional[str] = None, limit: int = 50, before: Optional[str] = None) -> dict:
        return self._c.request("GET", "/runs" + _q({"status": status, "agent": agent, "limit": limit, "before": before}))

    def get(self, run_id: str) -> dict: return self._c.request("GET", f"/runs/{run_id}")
    def pause(self, run_id: str) -> dict: return self._c.request("POST", f"/runs/{run_id}/pause")
    def resume(self, run_id: str) -> dict: return self._c.request("POST", f"/runs/{run_id}/resume")
    def cancel(self, run_id: str) -> dict: return self._c.request("POST", f"/runs/{run_id}/cancel")
    def retry(self, run_id: str) -> dict: return self._c.request("POST", f"/runs/{run_id}/retry")

    def approve(self, run_id: str, task_key: str, note: Optional[str] = None) -> dict:
        return self._c.request("POST", f"/runs/{run_id}/tasks/{_p(task_key)}/approve", {"note": note})

    def reject(self, run_id: str, task_key: str, note: Optional[str] = None) -> dict:
        return self._c.request("POST", f"/runs/{run_id}/tasks/{_p(task_key)}/reject", {"note": note})


class _Events(_Group):
    def list(self, run: Optional[str] = None, agent: Optional[str] = None, level: Optional[str] = None,
             q: Optional[str] = None, after: Optional[int] = None, limit: int = 100) -> List[dict]:
        return self._c.request("GET", "/events" + _q({"run": run, "agent": agent, "level": level, "q": q, "after": after, "limit": limit}))["events"]

    def stream(self, run: Optional[str] = None, after: Optional[int] = None, reconnect: bool = True) -> Iterator[dict]:
        """Yield live events (Server-Sent Events), reconnecting from the last id."""
        last = after
        while True:
            req = urllib.request.Request(self._c.base + "/events/stream" + _q({"run": run, "after": last}),
                                         headers={"Authorization": "Bearer " + self._c.api_key, "Accept": "text/event-stream"})
            try:
                with urllib.request.urlopen(req, timeout=70) as res:
                    data = []
                    for raw in res:
                        line = raw.decode("utf-8").rstrip("\n")
                        if line.startswith("data: "):
                            data.append(line[6:])
                        elif line == "" and data:
                            ev = json.loads("\n".join(data)); data = []
                            last = ev["id"]
                            yield ev
            except (urllib.error.URLError, TimeoutError):
                pass
            if not reconnect:
                return
            time.sleep(1)


class _Compute(_Group):
    def providers(self) -> List[dict]: return self._c.request("GET", "/compute/providers")["providers"]
    def add_provider(self, **p: Any) -> dict: return self._c.request("POST", "/compute/providers", p)
    def update_provider(self, provider_id: str, **patch: Any) -> dict: return self._c.request("PATCH", f"/compute/providers/{provider_id}", patch)
    def remove_provider(self, provider_id: str) -> None: self._c.request("DELETE", f"/compute/providers/{provider_id}")
    def test_provider(self, provider_id: str, model: Optional[str] = None) -> dict: return self._c.request("POST", f"/compute/providers/{provider_id}/test", {"model": model})
    def routing(self) -> dict: return self._c.request("GET", "/compute/routing")
    def set_routing(self, **r: Any) -> dict: return self._c.request("PUT", "/compute/routing", r)
    def route(self, model: str, policy: Optional[dict] = None) -> dict: return self._c.request("POST", "/compute/route", {"model": model, "policy": policy})
    def capacity(self) -> dict: return self._c.request("GET", "/compute/capacity")
    def models(self, q: Optional[str] = None) -> List[dict]: return self._c.request("GET", "/models" + _q({"q": q}))["data"]


class _Memory(_Group):
    def namespaces(self) -> List[dict]: return self._c.request("GET", "/memory/namespaces")["namespaces"]
    def create_namespace(self, name: str, kind: str = "long_term", **opts: Any) -> dict: return self._c.request("POST", "/memory/namespaces", {"name": name, "kind": kind, **opts})
    def update_namespace(self, ns: str, **patch: Any) -> dict: return self._c.request("PATCH", f"/memory/namespaces/{_p(ns)}", patch)
    def write(self, ns: str, content: Any, key: Optional[str] = None, metadata: Optional[dict] = None, ttl_seconds: Optional[int] = None) -> dict:
        body = {"content": content, "key": key, "metadata": metadata, "ttlSeconds": ttl_seconds}
        return self._c.request("POST", f"/memory/namespaces/{_p(ns)}/items", {k: v for k, v in body.items() if v is not None})
    def get(self, ns: str, key: str) -> dict: return self._c.request("GET", f"/memory/namespaces/{_p(ns)}/items" + _q({"key": key}))
    def list(self, ns: str, limit: int = 50) -> dict: return self._c.request("GET", f"/memory/namespaces/{_p(ns)}/items" + _q({"limit": limit}))
    def delete(self, ns: str, key_or_id: str) -> None: self._c.request("DELETE", f"/memory/namespaces/{_p(ns)}/items/{_p(key_or_id)}")
    def search(self, query: str, namespaces: Optional[List[str]] = None, k: int = 5, filter: Optional[dict] = None) -> List[dict]:
        body = {"query": query, "namespaces": namespaces, "k": k, "filter": filter}
        return self._c.request("POST", "/memory/search", {k2: v for k2, v in body.items() if v is not None})["results"]
    def usage(self) -> dict: return self._c.request("GET", "/memory/usage")


class ACP:
    def __init__(self, api_key: str, base_url: str = "https://gpu-sooty.vercel.app/api/v1", timeout: float = 60):
        if not api_key:
            raise ValueError("ACP: api_key is required")
        self.api_key, self.base, self.timeout = api_key, base_url.rstrip("/"), timeout
        self.agents, self.runs, self.events = _Agents(self), _Runs(self), _Events(self)
        self.compute, self.memory = _Compute(self), _Memory(self)

    def request(self, method: str, path: str, body: Any = None, headers: Optional[Dict[str, str]] = None) -> Any:
        data = json.dumps(body).encode() if body is not None else None
        h = {"Authorization": "Bearer " + self.api_key, "User-Agent": f"acp-python/{__version__}", **(headers or {})}
        if data is not None:
            h["Content-Type"] = "application/json"
        for attempt in range(4):
            req = urllib.request.Request(self.base + path, data=data, method=method, headers=h)
            try:
                with urllib.request.urlopen(req, timeout=self.timeout) as res:
                    raw = res.read()
                    return json.loads(raw) if raw else None
            except urllib.error.HTTPError as e:
                if (e.code == 429 or e.code >= 502) and attempt < 3:
                    time.sleep(0.5 * 2 ** attempt); continue
                try:
                    err = json.loads(e.read()).get("error", {})
                except Exception:
                    err = {}
                raise ACPError(e.code, err.get("code", "http_error"), err.get("message", str(e)), err.get("details")) from None
        raise ACPError(503, "unavailable", "retries exhausted")

    def wait_for_run(self, run_id: str, timeout: float = 300, interval: float = 1.0, until=TERMINAL) -> dict:
        deadline = time.time() + timeout
        while True:
            run = self.runs.get(run_id)
            if run["status"] in until:
                return run
            if time.time() > deadline:
                raise ACPError(408, "timeout", f"run {run_id} still {run['status']}")
            time.sleep(interval)

    def chat(self, messages: List[dict], model: str = "auto", **kw: Any) -> dict:
        """OpenAI-compatible chat completion through the ACP router."""
        return self.request("POST", "/chat/completions", {"model": model, "messages": messages, **kw})

    def chat_stream(self, messages: List[dict], model: str = "auto", **kw: Any) -> Iterator[str]:
        """Streamed chat completion: yields text fragments as they arrive."""
        body = json.dumps({"model": model, "messages": messages, **kw, "stream": True}).encode()
        req = urllib.request.Request(self.base + "/chat/completions", data=body, method="POST",
                                     headers={"Authorization": "Bearer " + self.api_key, "Content-Type": "application/json", "Accept": "text/event-stream"})
        try:
            res = urllib.request.urlopen(req, timeout=self.timeout)
        except urllib.error.HTTPError as e:
            err = json.loads(e.read() or b"{}").get("error", {})
            raise ACPError(e.code, err.get("code", "http_error"), err.get("message", str(e))) from None
        with res:
            for raw in res:
                line = raw.decode("utf-8").strip()
                if not line.startswith("data: ") or line == "data: [DONE]":
                    continue
                j = json.loads(line[6:])
                if "error" in j:
                    raise ACPError(502, j["error"].get("code", "upstream_failed"), j["error"].get("message", ""))
                delta = (j.get("choices") or [{}])[0].get("delta", {}).get("content")
                if delta:
                    yield delta

    def me(self) -> dict: return self.request("GET", "/me")
    def overview(self) -> dict: return self.request("GET", "/overview")
    def analytics(self, range: str = "24h") -> dict: return self.request("GET", "/analytics" + _q({"range": range}))
    def audit(self) -> dict: return self.request("GET", "/audit")
    def security_policy(self) -> dict: return self.request("GET", "/security/policy")
    def set_security_policy(self, **p: Any) -> dict: return self.request("PUT", "/security/policy", p)
    def create_key(self, name: str, scopes=("read", "write")) -> dict: return self.request("POST", "/keys", {"name": name, "scopes": list(scopes)})
    def revoke_key(self, key_id: str) -> None: self.request("DELETE", f"/keys/{key_id}")
    def create_webhook(self, url: str, events=("run.*",), description: str = "") -> dict: return self.request("POST", "/webhooks", {"url": url, "events": list(events), "description": description})
    def webhook_deliveries(self, webhook_id: str) -> dict: return self.request("GET", f"/webhooks/{webhook_id}/deliveries")


def verify_webhook(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    """Check an ACP-Signature header: t=<unix>,v1=<hex hmac_sha256(secret, "<t>.<body>")>."""
    try:
        parts = dict(p.split("=", 1) for p in header.split(","))
        t = int(parts["t"])
    except Exception:
        return False
    if abs(time.time() - t) > tolerance:
        return False
    body = raw_body.decode() if isinstance(raw_body, bytes) else raw_body
    expected = hmac.new(secret.encode(), f"{t}.{body}".encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))