#!/usr/bin/env python3
"""yakcloud dev 릴레이 에이전트 — 로컬(kind, NAT 뒤) 클러스터를 원격과 '동일한 콘솔'로.

콘솔 데이터소스 매니저 → yakcloud-api ds-actions 큐 → (이 에이전트가 롱폴로 수신) →
로컬 kind Secret 을 읽어 자격 해석 + 공유 도커 컨테이너의 격리 DB 에 접속 → 결과만 반환.
**자격(비밀번호)은 노트북을 절대 떠나지 않는다** — 경계엔 액션·소스식별자·고정 enum 결과만.

원격 백엔드(_ds_h_*)와 '동일한 응답 스키마'를 미러링한다(계약 보존) → 콘솔 매니저 무변경 재사용.
소스 접속은 컨테이너 네이티브 클라이언트(docker exec: psql/mysql/mongosh/redis-cli)로 —
노트북에 별도 드라이버 설치가 필요 없다("콘솔 하나로, 별도 툴 설치 없이").

환경변수(dev.py 가 주입):
  YAK_RELAY_KEY   릴레이 키(=콘솔 LOCAL cluster.id) — ds-actions 채널 키
  YAK_DS_TOKEN    per-cluster ds-token(Bearer) — 롱폴/결과 인증
  YAK_API_BASE    공개 yakcloud-api 베이스(예 https://api.yakenator.io)
  YAK_DEV_NS      소스 Secret 이 있는 네임스페이스(프로젝트명)
  YAK_DEV_CONTEXT kubectl 컨텍스트(기본 kind-yak-dev)
  YAK_DEV_RUNTIME docker|podman
  YAK_DEV_NET     공유 컨테이너 도커 네트워크(기본 kind)
"""
from __future__ import annotations

import base64
import hashlib
import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.request

CTX = os.environ.get("YAK_DEV_CONTEXT", "kind-yak-dev")
RUNTIME = os.environ.get("YAK_DEV_RUNTIME", "docker")
SHARED_NET = os.environ.get("YAK_DEV_NET", "kind")
SHARED_ADMIN = "yakdevadmin"
RELAY_KEY = os.environ.get("YAK_RELAY_KEY", "")
DS_TOKEN = os.environ.get("YAK_DS_TOKEN", "")
API_BASE = (os.environ.get("YAK_API_BASE", "") or "").rstrip("/")
NS = os.environ.get("YAK_DEV_NS", "")

_SHARED_CT = {"POSTGRESQL": "yak-dev-postgres", "MYSQL": "yak-dev-mysql", "MARIADB": "yak-dev-mariadb",
              "MONGODB": "yak-dev-mongo", "REDIS": "yak-dev-redis"}
_SVC_PORT = {"POSTGRESQL": 5432, "MYSQL": 3306, "MARIADB": 3306, "MONGODB": 27017, "REDIS": 6379}
_ROW_MAX = 500       # 백엔드 _DS_ROW_MAX 미러
_LIST_MAX = 500      # 백엔드 _DS_LIST_MAX 미러


class DsErr(Exception):
    """경계를 넘는 에러는 고정 enum code 만(드라이버 원문 미노출)."""
    def __init__(self, code: str):
        super().__init__(code)
        self.code = code


def log(m: str) -> None:
    print(f"\033[36m▸\033[0m [agent] {m}", flush=True)


# ── 로컬 자격 해석: kind Secret(yak-dev-src-*) → password/unit (자격은 여기서만) ──────────────
def _read_secret(service: str) -> dict:
    # 클러스터 전 네임스페이스에서 이름으로 Secret 조회(에이전트가 특정 프로젝트 ns 에 묶이지 않음).
    r = subprocess.run(["kubectl", "--context", CTX, "get", "secrets", "-A",
                        "--field-selector", f"metadata.name={service}", "-o", "json"],
                       capture_output=True, text=True)
    if r.returncode != 0:
        raise DsErr("NOT_FOUND")
    items = json.loads(r.stdout or "{}").get("items", []) or []
    if not items:
        raise DsErr("NOT_FOUND")
    data = (items[0].get("data", {}) or {})
    def dec(k: str) -> str:
        v = data.get(k)
        return base64.b64decode(v).decode() if v else ""
    return {"password": dec("password"), "unit": dec("unit")}


def _dbnum(unit: str) -> str:
    return str(int(hashlib.sha1(unit.encode()).hexdigest(), 16) % 16)


def _dexec(cname: str, argv: list[str], input_str: str | None = None) -> str:
    """docker/podman exec — argv 형태(셸 없음, 인젝션 안전). 실패=DRIVER_ERROR/UNREACHABLE."""
    r = subprocess.run([RUNTIME, "exec", "-i", cname, *argv], input=input_str,
                       capture_output=True, text=True)
    if r.returncode != 0:
        err = (r.stderr or "").lower()
        if "no such container" in err or "is not running" in err:
            raise DsErr("UNREACHABLE")
        raise DsErr("DRIVER_ERROR")
    return r.stdout


def _readonly_ok(sql: str) -> bool:  # 백엔드 _ds_readonly_ok 미러
    import re
    s = (sql or "").strip().rstrip(";").strip()
    if not s or ";" in s:
        return False
    return bool(re.match(r"(?is)^(select|with|show|explain|describe|desc|table)\b", s))


# ── SQL(pg/mysql/mariadb) — 격리 유저(unit)로 접속, 백엔드 _ds_h_sql 응답 스키마 미러 ──────────
def _h_sql(stype: str, sec: dict, action: str, params: dict) -> dict:
    cname = _SHARED_CT[stype]
    unit, pw = sec["unit"], sec["password"]
    is_pg = stype == "POSTGRESQL"
    client = "psql" if is_pg else ("mysql" if stype == "MYSQL" else "mariadb")

    def pg(sql: str, csv: bool = False) -> str:
        flags = ["--csv"] if csv else ["-tA"]
        return _dexec(cname, ["env", f"PGPASSWORD={pw}", "psql", "-U", unit, "-h", "127.0.0.1",
                              "-d", unit, "-v", "ON_ERROR_STOP=1", *flags, "-c", sql])

    def my(sql: str) -> str:  # --batch → TSV(헤더 포함), NULL=\N. 탭/개행은 mysql 이 이스케이프.
        return _dexec(cname, [client, "-u", unit, f"-p{pw}", "-h", "127.0.0.1", unit,
                              "--batch", "--raw", "-e", sql])

    def run_list(sql: str) -> list[str]:
        out = pg(sql) if is_pg else my(sql)
        rows = [ln for ln in out.splitlines() if ln != ""]
        if not is_pg and rows:
            rows = rows[1:]  # mysql --batch 헤더 제거
        return rows[:_LIST_MAX]

    scope = params.get("db") or ("public" if is_pg else unit)

    if action == "databases":
        if is_pg:
            sql = ("SELECT schema_name FROM information_schema.schemata "
                   "WHERE schema_name NOT LIKE 'pg_%' AND schema_name <> 'information_schema' ORDER BY 1")
        else:
            sql = ("SELECT schema_name FROM information_schema.schemata "
                   "WHERE schema_name NOT IN ('information_schema','mysql','performance_schema','sys') ORDER BY 1")
        return {"databases": run_list(sql)}

    if action in ("tables", "list", "collections"):
        if is_pg:
            names = run_list("SELECT table_name FROM information_schema.tables "
                             f"WHERE table_schema='{scope}' AND table_type='BASE TABLE' ORDER BY table_name")
            return {"db": scope, "tables": [{"name": n} for n in names]}
        out = my("SELECT table_name, table_rows FROM information_schema.tables "
                 f"WHERE table_schema='{scope}' AND table_type='BASE TABLE' ORDER BY table_name")
        rows = out.splitlines()[1:]
        tables = []
        for ln in rows[:_LIST_MAX]:
            c = ln.split("\t")
            tables.append({"name": c[0], "rows": int(c[1]) if len(c) > 1 and c[1].isdigit() else None})
        return {"db": scope, "tables": tables}

    if action == "columns":
        table = params["table"]
        sql = ("SELECT column_name, data_type, is_nullable FROM information_schema.columns "
               f"WHERE table_schema='{scope}' AND table_name='{table}' ORDER BY ordinal_position")
        out = pg(sql) if is_pg else my(sql)
        lines = out.splitlines()
        if not is_pg:
            lines = lines[1:]
        cols = []
        for ln in lines:
            if not ln:
                continue
            c = ln.split("\t" if not is_pg else "|")
            cols.append({"name": c[0], "type": c[1] if len(c) > 1 else "",
                         "nullable": (c[2].upper() == "YES") if len(c) > 2 else False})
        return {"db": scope, "table": table, "columns": cols}

    if action == "query":
        sql = (params.get("sql") or "").strip().rstrip(";").strip()
        if not _readonly_ok(sql):
            raise DsErr("FORBIDDEN_ACTION")
        limit = max(1, min(int(params.get("limit", 100)), _ROW_MAX))
        head = sql.lower()
        wrapped = f"SELECT * FROM ({sql}) AS _yak LIMIT {limit + 1}" if head.startswith(("select", "with")) else sql
        out = pg(wrapped, csv=True) if is_pg else my(wrapped)
        if is_pg:
            import csv as _csv
            import io as _io
            rr = list(_csv.reader(_io.StringIO(out)))
            colnames = rr[0] if rr else []
            data = rr[1:]
        else:
            lines = out.splitlines()
            colnames = lines[0].split("\t") if lines else []
            data = [ln.split("\t") for ln in lines[1:]]
        truncated = len(data) > limit
        return {"db": scope, "columns": colnames,
                "rows": [[None if v == ("" if is_pg else "\\N") else v for v in row] for row in data[:limit]],
                "rowCount": min(len(data), limit), "truncated": truncated}

    if action == "exec":
        s = (params.get("sql") or "").strip().rstrip(";").strip()
        if not s:
            raise DsErr("FORBIDDEN_ACTION")
        if is_pg:
            _dexec(cname, ["env", f"PGPASSWORD={pw}", "psql", "-U", unit, "-h", "127.0.0.1",
                           "-d", unit, "-v", "ON_ERROR_STOP=1", "-c", s])
        else:
            my(s)
        return {"db": scope, "rowsAffected": None}

    raise DsErr("FORBIDDEN_ACTION")


# ── MongoDB — 격리 유저로 mongosh, EJSON(canonical) 파싱. 읽기 코어(collections/find). ─────────
def _mongo_eval(sec: dict, js: str) -> str:
    unit, pw = sec["unit"], sec["password"]
    uri = f"mongodb://{unit}:{pw}@127.0.0.1:27017/{unit}"
    return _dexec(_SHARED_CT["MONGODB"], ["mongosh", uri, "--quiet", "--json=relaxed", "--eval", js])


def _h_mongo(sec: dict, action: str, params: dict) -> dict:
    db = sec["unit"]
    if action in ("collections", "list", "tables"):
        out = _mongo_eval(sec, "JSON.stringify(db.getCollectionNames())")
        names = json.loads(out or "[]")
        return {"db": db, "collections": [{"name": n} for n in names]}
    if action == "find":
        coll = params["coll"]
        filt = json.dumps(params.get("filter") or {})
        limit = max(1, min(int(params.get("limit", 50)), _ROW_MAX))
        skip = max(0, int(params.get("skip", 0)))
        js = (f"JSON.stringify(db.getCollection({json.dumps(coll)})"
              f".find({filt}).skip({skip}).limit({limit}).toArray())")
        docs = json.loads(_mongo_eval(sec, js) or "[]")
        return {"db": db, "coll": coll, "docs": docs, "total": len(docs), "skip": skip, "limit": limit}
    raise DsErr("FORBIDDEN_ACTION")


# ── Redis — 격리 DB 번호 + 공유 관리자 비번(dev). 읽기 코어(keys/get/dbsize). ──────────────────
def _redis(sec: dict, argv: list[str]) -> str:
    n = _dbnum(sec["unit"])
    return _dexec(_SHARED_CT["REDIS"], ["redis-cli", "-a", SHARED_ADMIN, "--no-auth-warning", "-n", n, *argv])


def _h_redis(sec: dict, action: str, params: dict) -> dict:
    if action == "keys":
        pattern = params.get("pattern") or "*"
        cursor = str(params.get("cursor", 0))
        out = _redis(sec, ["scan", cursor, "match", pattern, "count", "200"])
        lines = [ln for ln in out.splitlines() if ln != ""]
        nxt = int(lines[0]) if lines else 0
        keys = []
        for k in lines[1:]:
            t = _redis(sec, ["type", k]).strip()
            ttl = _redis(sec, ["ttl", k]).strip()
            keys.append({"name": k, "type": t, "ttl": int(ttl) if ttl.lstrip("-").isdigit() else -1})
        dbsize = _redis(sec, ["dbsize"]).strip()
        return {"cursor": nxt, "keys": keys[:_LIST_MAX], "dbsize": int(dbsize) if dbsize.isdigit() else 0}
    if action == "get":
        key = params["key"]
        t = _redis(sec, ["type", key]).strip()
        ttl = _redis(sec, ["ttl", key]).strip()
        if t == "string":
            val = _redis(sec, ["get", key]).rstrip("\n")
        elif t == "hash":
            val = _redis(sec, ["hgetall", key]).rstrip("\n")
        elif t in ("list",):
            val = _redis(sec, ["lrange", key, "0", "199"]).rstrip("\n")
        elif t == "set":
            val = _redis(sec, ["smembers", key]).rstrip("\n")
        elif t == "zset":
            val = _redis(sec, ["zrange", key, "0", "199", "withscores"]).rstrip("\n")
        else:
            val = ""
        return {"key": key, "type": t, "ttl": int(ttl) if ttl.lstrip("-").isdigit() else -1, "value": val}
    raise DsErr("FORBIDDEN_ACTION")


# ── 앱(워크로드) 라이브 조회 — 로컬 kind 에 kubectl(호스트). 원격 앱 화면 파리티(P2). ───────────
def _kubectl(argv: list[str]) -> str:
    # 클러스터 단위 에이전트 — NS 고정 안 함(호출자가 -A / --field-selector 로 네임스페이스 무관 조회).
    r = subprocess.run(["kubectl", "--context", CTX, *argv], capture_output=True, text=True)
    if r.returncode != 0:
        err = (r.stderr or "").lower()
        if "notfound" in err.replace(" ", "") or "not found" in err:
            raise DsErr("NOT_FOUND")
        raise DsErr("UNREACHABLE")
    return r.stdout


def _cpu_m(v) -> int:
    """CPU 수량 → millicores. '25m'→25, '1'→1000, '1.5'→1500."""
    if not v:
        return 0
    v = str(v)
    if v.endswith("m"):
        try:
            return int(float(v[:-1]))
        except ValueError:
            return 0
    if v.endswith("n"):  # nanocores(top 일부)
        return int(float(v[:-1]) / 1_000_000)
    try:
        return int(float(v) * 1000)
    except ValueError:
        return 0


def _mem_b(v) -> int:
    """메모리 수량 → bytes. Ki/Mi/Gi/Ti + K/M/G 접미사."""
    if not v:
        return 0
    v = str(v).strip()
    units = {"Ki": 1024, "Mi": 1024**2, "Gi": 1024**3, "Ti": 1024**4,
             "K": 1000, "M": 1000**2, "G": 1000**3, "T": 1000**4}
    for u, mult in units.items():
        if v.endswith(u):
            try:
                return int(float(v[:-len(u)]) * mult)
            except ValueError:
                return 0
    try:
        return int(float(v))
    except ValueError:
        return 0


def _h_kube(action: str, params: dict) -> dict:
    if action == "usage":
        nodes_j = json.loads(_kubectl(["get", "nodes", "-o", "json"]) or "{}").get("items", [])
        pods_j = json.loads(_kubectl(["get", "pods", "-A", "-o", "json"]) or "{}").get("items", [])
        # 노드별 예약(스케줄된 비종료 파드 컨테이너 requests 합)
        req = {}
        for p in pods_j:
            nn = (p.get("spec", {}) or {}).get("nodeName")
            if not nn or (p.get("status", {}) or {}).get("phase") in ("Succeeded", "Failed"):
                continue
            acc = req.setdefault(nn, [0, 0])
            for c in (p.get("spec", {}) or {}).get("containers", []) or []:
                rq = ((c.get("resources", {}) or {}).get("requests", {}) or {})
                acc[0] += _cpu_m(rq.get("cpu"))
                acc[1] += _mem_b(rq.get("memory"))
        # 실측(metrics-server 있으면): kubectl top nodes
        top = {}
        r = subprocess.run(["kubectl", "--context", CTX, "top", "nodes", "--no-headers"],
                           capture_output=True, text=True)
        if r.returncode == 0:
            for ln in r.stdout.splitlines():
                pt = ln.split()
                if len(pt) >= 5:  # NAME CPU(cores) CPU% MEM MEM%
                    top[pt[0]] = {"cpu": pt[1], "cpu_pct": pt[2].rstrip("%"), "mem": pt[3], "mem_pct": pt[4].rstrip("%")}
        out = []
        for n in nodes_j:
            meta, st = n.get("metadata", {}) or {}, n.get("status", {}) or {}
            name = meta.get("name")
            labels = meta.get("labels", {}) or {}
            roles = [k.split("/", 1)[1] for k in labels if k.startswith("node-role.kubernetes.io/")]
            cap, alloc = st.get("capacity", {}) or {}, st.get("allocatable", {}) or {}
            cpu_alloc = round(_cpu_m(alloc.get("cpu")) / 1000.0, 2)
            mem_alloc = round(_mem_b(alloc.get("memory")) / (1024**3), 2)
            rq = req.get(name, [0, 0])
            cpu_req = round(rq[0] / 1000.0, 2)
            mem_req = round(rq[1] / (1024**3), 2)
            ready = any(c.get("type") == "Ready" and c.get("status") == "True"
                        for c in (st.get("conditions", []) or []))
            t = top.get(name, {})
            # 디스크 실측 — kubelet Summary API(metrics-server 불필요, kind 지원). 원격(백엔드)과 동일 필드.
            disk = disk_pct = disk_cap = None
            ds = subprocess.run(["kubectl", "--context", CTX, "get", "--raw",
                                 f"/api/v1/nodes/{name}/proxy/stats/summary"], capture_output=True, text=True)
            if ds.returncode == 0:
                try:
                    fs = ((json.loads(ds.stdout).get("node", {}) or {}).get("fs", {})) or {}
                    used, capb = fs.get("usedBytes"), fs.get("capacityBytes")
                    if used is not None and capb:
                        disk = f"{round(used / 1024**3, 1)}Gi"
                        disk_cap = round(capb / 1024**3, 1)
                        disk_pct = str(round(used / capb * 100))
                except (ValueError, ZeroDivisionError, TypeError):
                    pass
            out.append({
                # kind 워커는 node-role 라벨이 없음 → role 없으면 worker(과거 control-plane fallback 은 워커를 CP 로 오분류).
                "name": name, "roles": roles or ["worker"], "joined": ready, "ready": ready,
                "pending": False, "maas_status": None,           # kind 노드 — MAAS 프로비저닝 개념 없음
                "phase": "Ready" if ready else "NotReady",        # NodeMetricDTO.phase(필수) — 상태 배지
                "cpu_cap": round(_cpu_m(cap.get("cpu")) / 1000.0, 2), "cpu_alloc": cpu_alloc,
                "cpu_req": cpu_req, "cpu_req_pct": round(cpu_req / cpu_alloc * 100) if cpu_alloc else None,
                "cpu": t.get("cpu"), "cpu_pct": t.get("cpu_pct"),
                "mem_cap": round(_mem_b(cap.get("memory")) / (1024**3), 2), "mem_alloc": mem_alloc,
                "mem_req": mem_req, "mem_req_pct": round(mem_req / mem_alloc * 100) if mem_alloc else None,
                "mem": t.get("mem"), "mem_pct": t.get("mem_pct"),
                "disk": disk, "disk_pct": disk_pct, "disk_cap": disk_cap,
            })
        return {"metrics_available": bool(top), "nodes": out,
                "autoscale": {"enabled": False, "min": None, "max": None},
                "provisioning": False, "history": []}
    if action == "pods":
        app = params.get("app") or ""
        sel = ["-l", f"app={app}"] if app else []
        items = json.loads(_kubectl(["get", "pods", "-A", *sel, "-o", "json"]) or "{}").get("items", [])
        pods = []
        for p in items:
            st = p.get("status", {}) or {}
            cs = st.get("containerStatuses", []) or []
            pods.append({"name": p["metadata"]["name"], "phase": st.get("phase", ""),
                         "ready": bool(cs) and all(c.get("ready") for c in cs),
                         "restarts": sum(int(c.get("restartCount", 0)) for c in cs),
                         "node": (p.get("spec", {}) or {}).get("nodeName", ""),
                         "ip": st.get("podIP", ""), "startedAt": st.get("startTime", "")})
        return {"app": app, "pods": pods[:100]}
    if action == "describe":
        pod = params["pod"]
        # 전 네임스페이스에서 파드 이름으로 조회(에이전트가 ns 무관).
        _items = json.loads(_kubectl(["get", "pods", "-A", "--field-selector", f"metadata.name={pod}", "-o", "json"]) or "{}").get("items", [])
        if not _items:
            raise DsErr("NOT_FOUND")
        p = _items[0]
        meta = p.get("metadata", {}) or {}
        spec = p.get("spec", {}) or {}
        st = p.get("status", {}) or {}
        cs = st.get("containerStatuses", []) or []
        spec_by_name = {c.get("name"): c for c in (spec.get("containers", []) or [])}
        restarts = sum(int(c.get("restartCount", 0)) for c in cs)
        containers = []
        for c in cs:
            sc = spec_by_name.get(c.get("name"), {}) or {}
            res = sc.get("resources", {}) or {}
            req, lim = res.get("requests", {}) or {}, res.get("limits", {}) or {}
            state = c.get("state", {}) or {}
            skey = next(iter(state.keys()), "")
            containers.append({
                "name": c.get("name", ""), "image": c.get("image"), "image_id": c.get("imageID"),
                "ready": bool(c.get("ready")), "restarts": int(c.get("restartCount", 0)),
                "state": skey, "state_detail": (state.get(skey, {}) or {}).get("reason"),
                "ports": [f"{pt.get('containerPort')}/{pt.get('protocol', 'TCP')}" for pt in (sc.get("ports", []) or [])],
                "cpu_request": req.get("cpu"), "mem_request": req.get("memory"),
                "cpu_limit": lim.get("cpu"), "mem_limit": lim.get("memory"),
                "liveness": "설정됨" if sc.get("livenessProbe") else None,
                "readiness": "설정됨" if sc.get("readinessProbe") else None,
                "env_from": [], "env": [],  # 민감할 수 있어 생략(MVP)
                "mounts": [{"path": m.get("mountPath", ""), "name": m.get("name", ""), "ro": bool(m.get("readOnly"))}
                           for m in (sc.get("volumeMounts", []) or [])],
            })
        owner = (meta.get("ownerReferences") or [{}])[0]
        return {
            "summary": {
                "name": meta.get("name"), "namespace": meta.get("namespace"), "node": spec.get("nodeName"),
                "phase": st.get("phase"), "pod_ip": st.get("podIP"), "host_ip": st.get("hostIP"),
                "qos_class": st.get("qosClass"), "service_account": spec.get("serviceAccountName"),
                "priority": spec.get("priority"),
                "controlled_by": f"{owner.get('kind')}/{owner.get('name')}" if owner.get("kind") else None,
                "start_time": st.get("startTime"), "restarts": restarts,
            },
            "labels": meta.get("labels", {}) or {}, "annotations": {},
            "containers": containers,
            "conditions": [{"type": c.get("type"), "status": c.get("status"), "reason": c.get("reason"),
                            "last_transition": c.get("lastTransitionTime")} for c in (st.get("conditions", []) or [])],
            "node_selector": spec.get("nodeSelector", {}) or {}, "tolerations": [],
            "volumes": [{"name": v.get("name", ""), "type": next(iter([k for k in v.keys() if k != "name"]), "")}
                        for v in (spec.get("volumes", []) or [])],
            "events": [],
        }
    if action == "workloads":
        items = json.loads(_kubectl(["get", "deploy", "-A", "-l", "app.kubernetes.io/managed-by=yakcloud-dev", "-o", "json"]) or "{}").get("items", [])
        wls = []
        for d in items:
            sp = d.get("spec", {}) or {}
            conts = (sp.get("template", {}).get("spec", {}) or {}).get("containers", []) or []
            wls.append({"name": d["metadata"]["name"], "replicas": sp.get("replicas", 0) or 0,
                        "ready": (d.get("status", {}) or {}).get("readyReplicas", 0) or 0,
                        "image": conts[0]["image"] if conts else ""})
        return {"workloads": wls}
    raise DsErr("FORBIDDEN_ACTION")


def dispatch(a: dict) -> dict:
    """ds-actions 액션 1건 실행 → {ok, data}|{ok:false, error:{code}} + nonce."""
    nonce = a.get("nonce")
    try:
        stype = str(a.get("type", "")).upper()
        service = a.get("service", "")
        action = a.get("action", "")
        params = a.get("params") or {}
        if stype == "KUBE":               # 앱(워크로드) 라이브 조회 — 소스 Secret 불필요
            return {"nonce": nonce, "ok": True, "data": _h_kube(action, params)}
        sec = _read_secret(service)
        if stype in ("POSTGRESQL", "MYSQL", "MARIADB"):
            data = _h_sql(stype, sec, action, params)
        elif stype == "MONGODB":
            data = _h_mongo(sec, action, params)
        elif stype == "REDIS":
            data = _h_redis(sec, action, params)
        else:
            raise DsErr("VALIDATION_FAILED")
        return {"nonce": nonce, "ok": True, "data": data}
    except DsErr as e:
        return {"nonce": nonce, "ok": False, "error": {"code": e.code}}
    except KeyError:
        return {"nonce": nonce, "ok": False, "error": {"code": "VALIDATION_FAILED"}}
    except Exception:  # noqa: BLE001 — 원문 미노출, 고정 enum 만
        return {"nonce": nonce, "ok": False, "error": {"code": "DRIVER_ERROR"}}


def _post_result(action_id: str, result: dict) -> None:
    body = json.dumps(result).encode()
    req = urllib.request.Request(f"{API_BASE}/clusters/{RELAY_KEY}/ds-actions/{action_id}/result",
                                 data=body, method="POST",
                                 headers={"Authorization": f"Bearer {DS_TOKEN}", "content-type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=15):
            pass
    except Exception as e:  # noqa: BLE001
        log(f"결과 전송 실패({action_id}): {e}")


def poll_once() -> int:
    """ds-actions 롱폴 1회(≤25s) → 받은 액션들 처리·결과 전송. 반환=처리 건수(-1=치명오류)."""
    req = urllib.request.Request(f"{API_BASE}/clusters/{RELAY_KEY}/ds-actions?wait=25",
                                 headers={"Authorization": f"Bearer {DS_TOKEN}"})
    try:
        with urllib.request.urlopen(req, timeout=35) as r:
            actions = (json.loads(r.read() or b"{}").get("actions") or [])
    except urllib.error.HTTPError as e:
        if e.code in (401, 403):
            log("ds-token 인증 실패(401/403) — 토큰 만료/폐기. 종료.")
            return -1
        time.sleep(3)
        return 0
    except Exception:  # noqa: BLE001 — 네트워크 일시 오류(노트북 슬립 등) → 재시도
        time.sleep(3)
        return 0
    for a in actions:
        result = dispatch(a)
        _post_result(a.get("id"), result)
        code = "" if result["ok"] else result["error"]["code"]
        log(f"{a.get('type')}·{a.get('action')} → {'ok' if result['ok'] else code}")
    return len(actions)


def main() -> None:
    if not (RELAY_KEY and DS_TOKEN and API_BASE):
        raise SystemExit("agent: YAK_RELAY_KEY/YAK_DS_TOKEN/YAK_API_BASE 필요")
    log(f"릴레이 연결(클러스터 단위): {API_BASE}/clusters/{RELAY_KEY}/ds-actions (ctx={CTX})")
    while True:
        if poll_once() < 0:
            break


if __name__ == "__main__":
    main()