ligbox-ops-platform/projects/ops-desk/api/app/agent_bindings.py

"""Persistência e enforcement — atribuições Agentics × função (Spec 027 UI-C)."""

from __future__ import annotations

import json
import sqlite3
from datetime import datetime, timezone
from typing import Any

RELATIONS = ("ui", "focus", "approve")
CAP_IDS = ("use_ui", "trigger_runs", "approve_runbooks", "configure_models")

LOCKED_AGENT_BINDINGS: frozenset[tuple[str, str, str]] = frozenset({
    ("A7", "agentic_operator", "approve"),
    ("A7", "super_admin", "approve"),
})

AGENT_IDS = ("A0", "A1", "A2", "A3", "A4", "A5", "A6", "A7")


def _spec():
    from app.rbac_matrix import AGENTIC_GOVERNANCE, AGENT_ROLE_MAP, ROLE_COLUMNS

    return AGENTIC_GOVERNANCE, AGENT_ROLE_MAP, ROLE_COLUMNS


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _default_agent_enabled(agent_id: str, role_id: str, relation: str) -> bool:
    _, AGENT_ROLE_MAP, _ = _spec()
    from app.rbac_matrix import AGENTIC_GOVERNANCE

    mapping = AGENT_ROLE_MAP.get(agent_id, {})
    if relation == "approve":
        return role_id in mapping.get("approvers", [])
    if relation == "focus":
        return role_id in mapping.get("operators", [])
    if relation == "ui":
        return role_id in AGENTIC_GOVERNANCE["use_ui"]
    return False


def _default_cap_enabled(cap_id: str, role_id: str) -> bool:
    AGENTIC_GOVERNANCE, _, _ = _spec()
    roles = AGENTIC_GOVERNANCE.get(cap_id)
    if isinstance(roles, list):
        return role_id in roles
    return False


def _is_locked(agent_id: str, role_id: str, relation: str) -> bool:
    return (agent_id, role_id, relation) in LOCKED_AGENT_BINDINGS


def init_schema(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS agent_role_bindings (
            agent_id TEXT NOT NULL,
            role_id TEXT NOT NULL,
            relation TEXT NOT NULL,
            enabled INTEGER NOT NULL DEFAULT 1,
            locked INTEGER NOT NULL DEFAULT 0,
            updated_at TEXT NOT NULL,
            updated_by TEXT,
            PRIMARY KEY (agent_id, role_id, relation)
        );
        CREATE TABLE IF NOT EXISTS agent_governance_caps (
            cap_id TEXT NOT NULL,
            role_id TEXT NOT NULL,
            enabled INTEGER NOT NULL DEFAULT 1,
            locked INTEGER NOT NULL DEFAULT 0,
            updated_at TEXT NOT NULL,
            updated_by TEXT,
            PRIMARY KEY (cap_id, role_id)
        );
        CREATE TABLE IF NOT EXISTS rbac_agent_audit (
            id INTEGER PRIMARY KEY,
            ts TEXT NOT NULL,
            username TEXT NOT NULL,
            action TEXT NOT NULL,
            payload_json TEXT NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_rbac_agent_audit_ts ON rbac_agent_audit(ts DESC);
        """
    )
    _seed_defaults(conn)


def _seed_defaults(conn: sqlite3.Connection) -> None:
    _, _, ROLE_COLUMNS = _spec()
    now = _now()
    if conn.execute("SELECT COUNT(*) c FROM agent_role_bindings").fetchone()["c"] == 0:
        rows = []
        for agent_id in AGENT_IDS:
            for role_id in ROLE_COLUMNS:
                for relation in RELATIONS:
                    enabled = 1 if _default_agent_enabled(agent_id, role_id, relation) else 0
                    locked = 1 if _is_locked(agent_id, role_id, relation) else 0
                    rows.append((agent_id, role_id, relation, enabled, locked, now, "system"))
        conn.executemany(
            """INSERT INTO agent_role_bindings
               (agent_id, role_id, relation, enabled, locked, updated_at, updated_by)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            rows,
        )
    if conn.execute("SELECT COUNT(*) c FROM agent_governance_caps").fetchone()["c"] == 0:
        cap_rows = []
        for cap_id in CAP_IDS:
            for role_id in ROLE_COLUMNS:
                enabled = 1 if _default_cap_enabled(cap_id, role_id) else 0
                cap_rows.append((cap_id, role_id, enabled, 0, now, "system"))
        conn.executemany(
            """INSERT INTO agent_governance_caps
               (cap_id, role_id, enabled, locked, updated_at, updated_by)
               VALUES (?, ?, ?, ?, ?, ?)""",
            cap_rows,
        )


def audit_log(
    conn: sqlite3.Connection,
    *,
    username: str,
    action: str,
    payload: dict[str, Any],
) -> None:
    conn.execute(
        "INSERT INTO rbac_agent_audit (ts, username, action, payload_json) VALUES (?, ?, ?, ?)",
        (_now(), username, action, json.dumps(payload)),
    )


def list_audit(conn: sqlite3.Connection, *, limit: int = 50) -> list[dict[str, Any]]:
    rows = conn.execute(
        "SELECT id, ts, username, action, payload_json FROM rbac_agent_audit ORDER BY id DESC LIMIT ?",
        (limit,),
    ).fetchall()
    out = []
    for r in rows:
        item = dict(r)
        try:
            item["payload"] = json.loads(item.pop("payload_json") or "{}")
        except json.JSONDecodeError:
            item["payload"] = {}
        out.append(item)
    return out


def _row_enabled(row: sqlite3.Row | None, default: bool = False) -> bool:
    if row is None:
        return default
    return bool(row["enabled"])


def agent_relations(conn: sqlite3.Connection, agent_id: str, role_id: str) -> list[str]:
    """Relações activas função ↔ agente."""
    if role_id == "super_admin":
        return list(RELATIONS)
    rels: list[str] = []
    for relation in RELATIONS:
        row = conn.execute(
            """SELECT enabled FROM agent_role_bindings
               WHERE agent_id=? AND role_id=? AND relation=?""",
            (agent_id, role_id, relation),
        ).fetchone()
        if _row_enabled(row, _default_agent_enabled(agent_id, role_id, relation)):
            rels.append(relation)
    return rels


def cap_enabled(conn: sqlite3.Connection, cap_id: str, role_id: str) -> bool:
    if role_id == "super_admin":
        return True
    row = conn.execute(
        "SELECT enabled FROM agent_governance_caps WHERE cap_id=? AND role_id=?",
        (cap_id, role_id),
    ).fetchone()
    return _row_enabled(row, _default_cap_enabled(cap_id, role_id))


def can_use_agentics_ui(conn: sqlite3.Connection, role_id: str) -> bool:
    if role_id == "super_admin":
        return True
    return cap_enabled(conn, "use_ui", role_id)


def can_chat_agent(conn: sqlite3.Connection, role_id: str, agent_id: str) -> bool:
    if role_id == "super_admin":
        return True
    rels = agent_relations(conn, agent_id, role_id)
    return bool(set(rels) & {"ui", "focus", "approve"})


def can_approve_agent(conn: sqlite3.Connection, role_id: str, agent_id: str) -> bool:
    if role_id == "super_admin":
        return True
    return "approve" in agent_relations(conn, agent_id, role_id)


def can_trigger_runs(conn: sqlite3.Connection, role_id: str) -> bool:
    if role_id == "super_admin":
        return True
    return cap_enabled(conn, "trigger_runs", role_id)


def can_approve_runbooks(conn: sqlite3.Connection, role_id: str) -> bool:
    if role_id == "super_admin":
        return True
    return cap_enabled(conn, "approve_runbooks", role_id)


def load_bindings_matrix(conn: sqlite3.Connection) -> dict[str, Any]:
    """Estado completo para UI / export."""
    agent_rows = conn.execute(
        "SELECT agent_id, role_id, relation, enabled, locked FROM agent_role_bindings"
    ).fetchall()
    cap_rows = conn.execute(
        "SELECT cap_id, role_id, enabled, locked FROM agent_governance_caps"
    ).fetchall()

    agents: dict[str, dict[str, dict[str, dict[str, Any]]]] = {}
    for r in agent_rows:
        agents.setdefault(r["agent_id"], {}).setdefault(r["role_id"], {})[r["relation"]] = {
            "enabled": bool(r["enabled"]),
            "locked": bool(r["locked"]),
        }

    caps: dict[str, dict[str, dict[str, Any]]] = {}
    for r in cap_rows:
        caps.setdefault(r["cap_id"], {})[r["role_id"]] = {
            "enabled": bool(r["enabled"]),
            "locked": bool(r["locked"]),
        }

    return {"agents": agents, "caps": caps}


def role_agentic_caps(conn: sqlite3.Connection, role_id: str) -> list[dict[str, str]]:
    from app.rbac_matrix import AGENTIC_GOVERNANCE

    caps = []
    for cap_id in CAP_IDS:
        if cap_enabled(conn, cap_id, role_id):
            caps.append({
                "id": cap_id,
                "label": AGENTIC_GOVERNANCE["labels"].get(cap_id, cap_id),
            })
    return caps


def set_agent_binding(
    conn: sqlite3.Connection,
    *,
    agent_id: str,
    role_id: str,
    relation: str,
    enabled: bool,
    username: str,
) -> dict[str, Any]:
    _, _, ROLE_COLUMNS = _spec()
    if agent_id not in AGENT_IDS:
        raise ValueError(f"agente inválido: {agent_id}")
    if role_id not in ROLE_COLUMNS:
        raise ValueError(f"função inválida: {role_id}")
    if relation not in RELATIONS:
        raise ValueError(f"relação inválida: {relation}")

    row = conn.execute(
        """SELECT enabled, locked FROM agent_role_bindings
           WHERE agent_id=? AND role_id=? AND relation=?""",
        (agent_id, role_id, relation),
    ).fetchone()
    locked = bool(row["locked"]) if row else _is_locked(agent_id, role_id, relation)
    if locked and not enabled:
        raise ValueError("binding bloqueado pela Spec 027 (obrigatório)")

    now = _now()
    if row:
        conn.execute(
            """UPDATE agent_role_bindings SET enabled=?, updated_at=?, updated_by=?
               WHERE agent_id=? AND role_id=? AND relation=?""",
            (1 if enabled else 0, now, username, agent_id, role_id, relation),
        )
    else:
        conn.execute(
            """INSERT INTO agent_role_bindings
               (agent_id, role_id, relation, enabled, locked, updated_at, updated_by)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (agent_id, role_id, relation, 1 if enabled else 0, 1 if locked else 0, now, username),
        )

    audit_log(
        conn,
        username=username,
        action="agent_binding.update",
        payload={
            "agent_id": agent_id,
            "role_id": role_id,
            "relation": relation,
            "enabled": enabled,
        },
    )
    return {
        "agent_id": agent_id,
        "role_id": role_id,
        "relation": relation,
        "enabled": enabled,
        "locked": locked,
    }


def set_governance_cap(
    conn: sqlite3.Connection,
    *,
    cap_id: str,
    role_id: str,
    enabled: bool,
    username: str,
) -> dict[str, Any]:
    _, _, ROLE_COLUMNS = _spec()
    if cap_id not in CAP_IDS:
        raise ValueError(f"capacidade inválida: {cap_id}")
    if role_id not in ROLE_COLUMNS:
        raise ValueError(f"função inválida: {role_id}")

    row = conn.execute(
        "SELECT enabled, locked FROM agent_governance_caps WHERE cap_id=? AND role_id=?",
        (cap_id, role_id),
    ).fetchone()
    locked = bool(row["locked"]) if row else False
    if locked and not enabled:
        raise ValueError("capacidade bloqueada")

    now = _now()
    if row:
        conn.execute(
            """UPDATE agent_governance_caps SET enabled=?, updated_at=?, updated_by=?
               WHERE cap_id=? AND role_id=?""",
            (1 if enabled else 0, now, username, cap_id, role_id),
        )
    else:
        conn.execute(
            """INSERT INTO agent_governance_caps
               (cap_id, role_id, enabled, locked, updated_at, updated_by)
               VALUES (?, ?, ?, 0, ?, ?)""",
            (cap_id, role_id, 1 if enabled else 0, now, username),
        )

    audit_log(
        conn,
        username=username,
        action="agent_cap.update",
        payload={"cap_id": cap_id, "role_id": role_id, "enabled": enabled},
    )
    return {"cap_id": cap_id, "role_id": role_id, "enabled": enabled, "locked": locked}