oclaw/runtime/skill_role_binding.py

from __future__ import annotations

import json
import os
from typing import Any

from runtime.agents.specialists import discover_specialist_ids

SKILL_ROLE_BINDING_KEY = "skill_role_binding"
SKILL_ROLE_BINDING_ENABLED_SETTING = "AIA_SKILL_ROLE_BINDING_ENABLED"
# Deprecated: Manager inherit removed; kept so old settings keys do not crash readers.
SKILL_ROLE_BINDING_MANAGER_INHERIT_SETTING = "AIA_SKILL_ROLE_BINDING_MANAGER_INHERIT"


def _truthy(v: str | None) -> bool:
    return str(v or "").strip().lower() in {"1", "true", "yes", "on"}


def ordered_specialist_ids() -> list[str]:
    base = [str(k).strip().lower() for k in discover_specialist_ids() if str(k).strip()]
    preferred = [x for x in ("generalist", "ops", "memory") if x in set(base)]
    return preferred + [x for x in base if x not in set(preferred)]


def ordered_binding_roles() -> list[str]:
    """Roles that appear in skill-binding UI/maps (specialists only; no manager)."""
    return ordered_specialist_ids()


def skill_role_binding_enabled_env_present() -> bool:
    """True when ``AIA_SKILL_ROLE_BINDING_ENABLED`` is set in the process environment (any value)."""
    return bool(str(os.getenv(SKILL_ROLE_BINDING_ENABLED_SETTING) or "").strip())


def skill_role_binding_enabled_stored(*, store: Any) -> bool:
    """SQLite/Admin value only; ignores environment (contrast :func:`skill_role_binding_enabled`)."""
    try:
        raw = str(store.get_setting(SKILL_ROLE_BINDING_ENABLED_SETTING) or "").strip()
    except Exception:
        raw = ""
    return _truthy(raw) if raw else False


def skill_role_binding_enabled(*, store: Any) -> bool:
    raw_env = str(os.getenv(SKILL_ROLE_BINDING_ENABLED_SETTING) or "").strip()
    if raw_env:
        return _truthy(raw_env)
    try:
        raw = str(store.get_setting(SKILL_ROLE_BINDING_ENABLED_SETTING) or "").strip()
    except Exception:
        raw = ""
    return _truthy(raw) if raw else False


def load_skill_role_binding_dict(store: Any) -> dict[str, Any]:
    try:
        raw = str(store.get_setting(SKILL_ROLE_BINDING_KEY) or "").strip()
    except Exception:
        raw = ""
    if not raw:
        return {}
    try:
        obj = json.loads(raw)
    except Exception:
        return {}
    return obj if isinstance(obj, dict) else {}


def mapping_has_any_skill_names(mapping: dict[str, list[str]]) -> bool:
    for vals in mapping.values():
        if vals and any(str(x).strip() for x in vals):
            return True
    return False


def normalize_skill_role_binding(
    *,
    mapping_raw: dict[str, Any],
    valid_skill_names: set[str],
    available_roles: list[str] | None = None,
) -> dict[str, list[str]]:
    roles = available_roles if available_roles is not None else ordered_binding_roles()
    role_set = set(roles)
    out: dict[str, list[str]] = {r: [] for r in roles}

    def _append(rk: str, items: Any) -> None:
        if rk not in role_set:
            return
        rows = items if isinstance(items, list) else []
        seen = set(out[rk])
        for x in rows:
            nm = str(x or "").strip()
            if not nm or nm not in valid_skill_names or nm in seen:
                continue
            seen.add(nm)
            out[rk].append(nm)

    for k, v in (mapping_raw or {}).items():
        rk = str(k or "").strip().lower()
        # Legacy Manager bindings fold into generalist.
        if rk == "manager":
            rk = "generalist"
        _append(rk, v)
    return out


def allowed_workspace_skill_names_for_role(*, store: Any, role: str) -> set[str]:
    """Skills bound to the given specialist role, plus public workspace skills."""
    r = str(role or "").strip().lower()
    if r == "manager":
        r = "generalist"
    if not r:
        return set()
    from runtime.skills import discover_public_workspace_skill_names

    public = discover_public_workspace_skill_names()
    mapping = normalize_skill_role_binding(
        mapping_raw=load_skill_role_binding_dict(store),
        valid_skill_names=_all_installed_skill_names(store),
    )
    sp = {str(x).strip() for x in (mapping.get(r) or []) if str(x).strip()}
    return sp | public


def _all_installed_skill_names(store: Any) -> set[str]:
    from runtime.skills import discover_workspace_skill_manifests

    return {str(m.name).strip() for m in discover_workspace_skill_manifests() if str(m.name or "").strip()}


def should_apply_workspace_role_filter(*, store: Any, skill_binding_role: str | None) -> bool:
    """When role binding is enabled, always filter the workspace skill catalog by role.

    Empty binding maps still apply: each role then only sees ``public`` workspace skills
    (see :func:`allowed_workspace_skill_names_for_role`), not the full install tree.
    """
    if not str(skill_binding_role or "").strip():
        return False
    if not skill_role_binding_enabled(store=store):
        return False
    return True


__all__ = [
    "SKILL_ROLE_BINDING_KEY",
    "SKILL_ROLE_BINDING_ENABLED_SETTING",
    "SKILL_ROLE_BINDING_MANAGER_INHERIT_SETTING",
    "skill_role_binding_enabled_env_present",
    "skill_role_binding_enabled_stored",
    "allowed_workspace_skill_names_for_role",
    "load_skill_role_binding_dict",
    "mapping_has_any_skill_names",
    "normalize_skill_role_binding",
    "ordered_binding_roles",
    "ordered_specialist_ids",
    "should_apply_workspace_role_filter",
    "skill_role_binding_enabled",
]