| Your priority            | What CUGA gives you                                          |
| ------------------------ | ------------------------------------------------------------ |
| **Sovereignty**          | Run the harness in your environment. Connect privately hosted models and tools. Preload supporting model artifacts for disconnected deployments. |
| **Execution policies**   | Block disallowed intents, require approval for selected tools, guide workflows with playbooks, and configure tool-level guards. |
| **Inference efficiency** | Choose open models, tune reasoning modes, shortlist relevant tools, set tool-call budgets, and inspect token usage with per-run receipts. |
| **Faster delivery**      | Connect MCP servers, OpenAPI services, and Python tools. Configure and test an agent in the UI before publishing a version. |
| **Recurring automation** | Use the separate events service to run agents on schedules, webhooks, and application events, with run history in Events Studio. |
| **Reusable experience**  | Load task-specific skills and use a configured Evolve integration to retrieve relevant guidance and user preferences. |

"""Synthetic CRM data and the in-memory store the demo tools operate on.

All data is fictional and local, so the showcase has no cloud dependency.
"""

from __future__ import annotations

import copy
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

# Account records. ``usage`` is monthly active seats for the last six months.
_ACCOUNTS: Dict[str, Dict[str, Any]] = {
    "A-100": {
        "id": "A-100", "name": "Acme Corp", "plan": "Enterprise", "arr": 240000,
        "renewal_date": "2026-12-01", "nps": 52, "owner": "Dana Whitfield",
        "usage": [820, 845, 890, 910, 960, 1010],
    },
    "A-200": {
        "id": "A-200", "name": "Globex", "plan": "Pro", "arr": 96000,
        "renewal_date": "2026-11-15", "nps": 18, "owner": "Marcus Lee",
        "usage": [700, 660, 610, 540, 500, 455],
    },
    "A-300": {
        "id": "A-300", "name": "Initech", "plan": "Starter", "arr": 18000,
        "renewal_date": "2027-02-01", "nps": 35, "owner": "Priya Nair",
        "usage": [200, 201, 199, 202, 200, 203],
    },
    "A-400": {
        "id": "A-400", "name": "Umbrella", "plan": "Enterprise", "arr": 410000,
        "renewal_date": "2026-12-20", "nps": 41, "owner": "Dana Whitfield",
        "usage": [1500, 1490, 1470, 1460, 1440, 1430],
    },
}

_TICKETS: List[Dict[str, Any]] = [
    {"id": "T-1", "account_id": "A-100", "priority": "P3", "title": "Dashboard colour palette request", "status": "open"},
    {"id": "T-2", "account_id": "A-200", "priority": "P1", "title": "API outage in EU region", "status": "open"},
    {"id": "T-3", "account_id": "A-200", "priority": "P2", "title": "Export job times out", "status": "open"},
    {"id": "T-4", "account_id": "A-200", "priority": "P2", "title": "SSO login loop", "status": "open"},
    {"id": "T-5", "account_id": "A-400", "priority": "P2", "title": "Slow report generation", "status": "open"},
    {"id": "T-6", "account_id": "A-300", "priority": "P3", "title": "Invoice address change", "status": "closed"},
]

# Health score penalties per open ticket priority.
_TICKET_PENALTY = {"P1": 15, "P2": 6, "P3": 2}


def usage_trend_pct(usage: List[int]) -> float:
    """Percent change of the last three months vs the three months before."""
    if len(usage) < 6:
        return 0.0
    prev = sum(usage[-6:-3]) / 3
    last = sum(usage[-3:]) / 3
    return round((last - prev) / prev * 100, 1) if prev else 0.0


def health_label(score: int) -> str:
    """Map a numeric score to a text label (never colour alone)."""
    if score >= 75:
        return "healthy"
    if score >= 50:
        return "needs attention"
    return "at risk"


def compute_health(account: Dict[str, Any], tickets: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Deterministic account health score in the 0-100 range.

    score = 100 - ticket penalties + trend bonus (capped +-15) + NPS adjustment
    """
    penalty = sum(_TICKET_PENALTY.get(t["priority"], 0) for t in tickets if t["status"] == "open")
    trend = usage_trend_pct(account["usage"])
    trend_bonus = max(-15.0, min(15.0, trend * 0.8))
    nps_adj = (account["nps"] - 30) * 0.2
    score = int(round(max(0.0, min(100.0, 100 - penalty + trend_bonus + nps_adj - 10))))
    return {
        "account_id": account["id"],
        "score": score,
        "label": health_label(score),
        "drivers": {
            "ticket_penalty": penalty,
            "usage_trend_pct": trend,
            "nps": account["nps"],
        },
    }


@dataclass
class CrmStore:
    """Mutable in-memory CRM. One instance per browser session."""

    accounts: Dict[str, Dict[str, Any]] = field(default_factory=lambda: copy.deepcopy(_ACCOUNTS))
    tickets: List[Dict[str, Any]] = field(default_factory=lambda: copy.deepcopy(_TICKETS))
    outbox: List[Dict[str, Any]] = field(default_factory=list)
    deleted: List[str] = field(default_factory=list)

    def find_account(self, query: str) -> Optional[Dict[str, Any]]:
        """Find an account by id or (partial, case-insensitive) name."""
        q = (query or "").strip().lower()
        if not q:
            return None
        for acc in self.accounts.values():
            if acc["id"].lower() == q or acc["name"].lower() == q:
                return acc
        for acc in self.accounts.values():
            if q in acc["name"].lower():
                return acc
        return None

    def open_tickets(self, account_id: str) -> List[Dict[str, Any]]:
        return [t for t in self.tickets if t["account_id"] == account_id and t["status"] == "open"]

    def reset(self) -> None:
        """Restore the pristine demo data."""
        self.accounts = copy.deepcopy(_ACCOUNTS)
        self.tickets = copy.deepcopy(_TICKETS)
        self.outbox.clear()
        self.deleted.clear()

{"ok": True,  "data": {...}}
{"ok": False, "code": "not_found", "error": "No account matches 'Foo'."}

"""Demo tools shared by the offline simulator and the live ``CugaAgent``.

Each tool is a plain function returning the *envelope* convention used by the
official cuga-apps reference apps::

    {"ok": True,  "data": ...}
    {"ok": False, "code": "not_found", "error": "..."}

``ToolSpec`` carries the metadata (description, sensitivity, owning sub-agent)
so the same registry can be exposed as LangChain tools in Live mode.
"""

from __future__ import annotations

import ast
import operator
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional

from .demo_data import CrmStore, compute_health, usage_trend_pct

Envelope = Dict[str, Any]


def ok(data: Any) -> Envelope:
    """Success envelope."""
    return {"ok": True, "data": data}


def err(code: str, message: str) -> Envelope:
    """Failure envelope."""
    return {"ok": False, "code": code, "error": message}


# --------------------------------------------------------------------------
# Safe calculator (no eval): walks the AST and allows arithmetic only.
# --------------------------------------------------------------------------
_BIN_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.Mod: operator.mod,
}
_UNARY_OPS = {ast.UAdd: operator.pos, ast.USub: operator.neg}


def _eval_node(node: ast.AST) -> float:
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)) and not isinstance(node.value, bool):
        return node.value
    if isinstance(node, ast.BinOp) and type(node.op) in _BIN_OPS:
        left, right = _eval_node(node.left), _eval_node(node.right)
        if isinstance(node.op, ast.Pow) and abs(right) > 12:
            raise ValueError("exponent too large")
        return _BIN_OPS[type(node.op)](left, right)
    if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPS:
        return _UNARY_OPS[type(node.op)](_eval_node(node.operand))
    raise ValueError("unsupported expression")


def calculator(expression: str) -> Envelope:
    """Evaluate an arithmetic expression such as ``(240000 + 96000) * 0.1``."""
    try:
        tree = ast.parse((expression or "").strip(), mode="eval")
        value = _eval_node(tree.body)
    except ZeroDivisionError:
        return err("math_error", "Division by zero.")
    except (ValueError, SyntaxError):
        return err("bad_expression", "Only plain arithmetic is supported.")
    return ok({"expression": expression, "value": round(value, 6)})


# --------------------------------------------------------------------------
# CRM tools (bound to a CrmStore through ``build_registry``)
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class ToolSpec:
    """Metadata + callable for one tool."""

    name: str
    description: str
    fn: Callable[..., Envelope]
    sensitive: bool = False  # sensitive tools are the usual approval candidates
    agent: str = "cuga"  # owning sub-agent in supervisor mode
    params: Optional[Dict[str, str]] = None  # param name -> short description

    def call(self, **kwargs: Any) -> Envelope:
        """Invoke the tool, converting unexpected exceptions into envelopes."""
        try:
            return self.fn(**kwargs)
        except TypeError as exc:  # wrong/missing arguments
            return err("bad_arguments", str(exc))
        except Exception as exc:  # noqa: BLE001 - tools must never crash a run
            return err("tool_error", str(exc))


def build_registry(
    store: CrmStore, knowledge_search: Optional[Callable[[str], Envelope]] = None
) -> Dict[str, ToolSpec]:
    """Create the tool registry bound to ``store``.

    Args:
        store: The session's CRM store.
        knowledge_search: Optional callable backing ``search_knowledge``.
    """

    def get_account(query: str) -> Envelope:
        acc = store.find_account(query)
        if not acc:
            return err("not_found", f"No account matches '{query}'.")
        return ok({k: v for k, v in acc.items() if k != "usage"})

    def list_open_tickets(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok(store.open_tickets(acc["id"]))

    def compute_health_score(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok(compute_health(acc, store.open_tickets(acc["id"])))

    def get_usage_trend(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok({"account_id": acc["id"], "monthly_seats": acc["usage"], "trend_pct": usage_trend_pct(acc["usage"])})

    def list_accounts() -> Envelope:
        return ok([{"id": a["id"], "name": a["name"], "plan": a["plan"], "arr": a["arr"]} for a in store.accounts.values()])

    def send_email(to: str, subject: str, body: str) -> Envelope:
        if "@" not in (to or ""):
            return err("bad_arguments", "Recipient must be an email address.")
        store.outbox.append({"to": to, "subject": subject, "body": body})
        return ok({"queued": True, "to": to, "subject": subject})

    def delete_account(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        store.accounts.pop(acc["id"])
        store.deleted.append(acc["id"])
        return ok({"deleted": acc["id"], "name": acc["name"]})

    def search_knowledge(query: str) -> Envelope:
        if knowledge_search is None:
            return err("unavailable", "Knowledge base is not configured.")
        return knowledge_search(query)

    specs = [
        ToolSpec("list_accounts", "List all CRM accounts.", list_accounts, agent="crm_agent", params={}),
        ToolSpec("get_account", "Look up one account by id or name.", get_account, agent="crm_agent",
                 params={"query": "Account id (A-100) or name (Acme)"}),
        ToolSpec("get_usage_trend", "Monthly active seats and 3-month trend for an account.", get_usage_trend,
                 agent="crm_agent", params={"account_id": "Account id or name"}),
        ToolSpec("list_open_tickets", "Open support tickets for an account.", list_open_tickets, agent="support_agent",
                 params={"account_id": "Account id or name"}),
        ToolSpec("compute_health_score", "Deterministic 0-100 health score with drivers.", compute_health_score,
                 agent="support_agent", params={"account_id": "Account id or name"}),
        ToolSpec("calculator", "Evaluate an arithmetic expression.", calculator, agent="crm_agent",
                 params={"expression": "Arithmetic such as 240000*0.1"}),
        ToolSpec("search_knowledge", "Search the local knowledge base and return cited passages.", search_knowledge,
                 agent="crm_agent", params={"query": "Natural-language question"}),
        ToolSpec("send_email", "Queue an email to a contact.", send_email, sensitive=True, agent="comms_agent",
                 params={"to": "Recipient address", "subject": "Subject line", "body": "Message body"}),
        ToolSpec("delete_account", "Permanently delete an account.", delete_account, sensitive=True,
                 agent="crm_agent", params={"account_id": "Account id or name"}),
    ]
    return {s.name: s for s in specs}


def tool_names(registry: Dict[str, ToolSpec]) -> List[str]:
    """Stable, sorted tool names."""
    return sorted(registry)

"""Policy engine for the offline simulator.

It reproduces, in plain Python, the *behaviour* of CUGA's five policy types so
users can experiment without an LLM:

* ``intent_guard``     - blocks a request before any planning happens
* ``playbook``         - injects step-by-step guidance when a request matches
* ``tool_guide``       - enriches a tool's description with extra guidance
* ``tool_approval``    - parks a tool call until a human approves it (HITL)
* ``output_formatter`` - reshapes the final answer

In Live mode the very same definitions are registered on the real
``agent.policies`` manager (see ``live.py``).
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from typing import Dict, List, Optional

POLICY_TYPES = ("intent_guard", "playbook", "tool_guide", "tool_approval", "output_formatter")

TYPE_LABELS = {
    "intent_guard": "Intent Guard",
    "playbook": "Playbook",
    "tool_guide": "Tool Guide",
    "tool_approval": "Tool Approval",
    "output_formatter": "Output Formatter",
}

TYPE_HELP = {
    "intent_guard": "Blocks a request that matches keywords before the agent plans anything.",
    "playbook": "Adds step-by-step guidance when the request matches keywords.",
    "tool_guide": "Appends usage guidance to the descriptions of the target tools.",
    "tool_approval": "Pauses before the listed tools run and asks a human to approve.",
    "output_formatter": "Rewrites the final answer using a format instruction.",
}


@dataclass
class Policy:
    """One policy definition (superset of the fields of the five SDK types)."""

    name: str
    type: str
    keywords: List[str] = field(default_factory=list)
    content: str = ""  # playbook / tool_guide content, formatter instruction
    response: str = ""  # intent_guard response
    tools: List[str] = field(default_factory=list)  # tool_approval / tool_guide targets
    message: str = ""  # tool_approval prompt
    enabled: bool = True
    id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])

    def matches(self, text: str) -> bool:
        """Case-insensitive keyword OR-match (the SDK's default operator)."""
        lowered = (text or "").lower()
        return any(k.strip().lower() in lowered for k in self.keywords if k.strip())


def default_policies() -> List[Policy]:
    """Starter policy set shown in the Policy Lab."""
    return [
        Policy(
            name="Block salary disclosure", type="intent_guard",
            keywords=["salary", "compensation of", "payroll"],
            response="I can't share compensation data. Please contact HR.",
        ),
        Policy(
            name="Churn-risk playbook", type="playbook",
            keywords=["churn", "at risk", "renewal risk"],
            content="1. Fetch the account.\n2. Check open P1/P2 tickets.\n3. Compare the 3-month usage trend.\n4. Recommend one concrete next action.",
        ),
        Policy(
            name="Email etiquette", type="tool_guide", tools=["send_email"],
            content="Keep emails under 120 words. Always include the account owner in the greeting.",
        ),
        Policy(
            name="Approve outbound & destructive actions", type="tool_approval",
            tools=["send_email", "delete_account"],
            message="This action changes data outside the conversation. Approve?",
        ),
        Policy(
            name="Executive summary format", type="output_formatter",
            keywords=[], content="Answer as a bold one-line verdict followed by up to three bullet points.",
            enabled=False,
        ),
    ]


class PolicySet:
    """A mutable collection of policies with the lookups the simulator needs."""

    def __init__(self, policies: Optional[List[Policy]] = None) -> None:
        self._items: Dict[str, Policy] = {}
        for p in policies if policies is not None else default_policies():
            self._items[p.id] = p

    # -- CRUD ---------------------------------------------------------------
    def add(self, policy: Policy) -> Policy:
        if policy.type not in POLICY_TYPES:
            raise ValueError(f"Unknown policy type: {policy.type}")
        if not policy.name.strip():
            raise ValueError("Policy name is required.")
        self._items[policy.id] = policy
        return policy

    def remove(self, policy_id: str) -> bool:
        return self._items.pop(policy_id, None) is not None

    def clear(self) -> None:
        self._items.clear()

    def toggle(self, policy_id: str, enabled: bool) -> None:
        if policy_id in self._items:
            self._items[policy_id].enabled = enabled

    def all(self) -> List[Policy]:
        return list(self._items.values())

    def active(self, ptype: str) -> List[Policy]:
        return [p for p in self._items.values() if p.type == ptype and p.enabled]

    # -- lookups used by the agent ------------------------------------------
    def intent_guard(self, message: str) -> Optional[Policy]:
        return next((p for p in self.active("intent_guard") if p.matches(message)), None)

    def playbooks(self, message: str) -> List[Policy]:
        return [p for p in self.active("playbook") if p.matches(message)]

    def tool_guides(self, tool: str) -> List[Policy]:
        return [p for p in self.active("tool_guide") if tool in p.tools or "*" in p.tools]

    def approval_for(self, tool: str) -> Optional[Policy]:
        return next((p for p in self.active("tool_approval") if tool in p.tools or "*" in p.tools), None)

    def formatter(self) -> Optional[Policy]:
        return next(iter(self.active("output_formatter")), None)

"""Loader for ``SKILL.md`` files (the format used by cuga-skills).

A skill is a Markdown file with YAML-style front matter (``name``,
``description``, ``tags``) followed by instructions. CUGA loads skills from
``.cuga/skills``; here we also match them to requests in the simulator.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple

from .knowledge import tokenize


@dataclass(frozen=True)
class Skill:
    name: str
    description: str
    body: str
    tags: Tuple[str, ...] = field(default_factory=tuple)
    path: str = ""

    def score(self, message: str) -> float:
        """Share of the description's keywords found in ``message`` (0-1)."""
        wanted = set(tokenize(self.description))
        if not wanted:
            return 0.0
        return len(wanted & set(tokenize(message))) / len(wanted)


def parse_skill(text: str, path: str = "") -> Optional[Skill]:
    """Parse one SKILL.md; returns None when the front matter is invalid."""
    match = re.match(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", text, re.DOTALL)
    if not match:
        return None
    meta: Dict[str, object] = {}
    current_list: Optional[str] = None
    for line in match.group(1).splitlines():
        if re.match(r"^\s+-\s+", line) and current_list:
            meta.setdefault(current_list, []).append(re.sub(r"^\s+-\s+", "", line).strip())  # type: ignore[union-attr]
            continue
        kv = re.match(r"^([A-Za-z_]+):\s*(.*)$", line)
        if kv:
            key, value = kv.group(1), kv.group(2).strip()
            if value:
                meta[key] = value
                current_list = None
            else:
                meta[key] = []
                current_list = key
    name, description = meta.get("name"), meta.get("description")
    if not isinstance(name, str) or not isinstance(description, str):
        return None
    tags = tuple(meta.get("tags", [])) if isinstance(meta.get("tags"), list) else ()
    return Skill(name=name, description=description, body=match.group(2).strip(), tags=tags, path=path)


def load_skills(folder: Path) -> List[Skill]:
    """Load every ``*/SKILL.md`` below ``folder`` (invalid files are skipped)."""
    skills: List[Skill] = []
    if not folder.is_dir():
        return skills
    for path in sorted(folder.rglob("SKILL.md")):
        try:
            skill = parse_skill(path.read_text(encoding="utf-8"), str(path))
        except (OSError, UnicodeDecodeError):
            continue
        if skill:
            skills.append(skill)
    return skills


def match_skill(skills: List[Skill], message: str, threshold: float = 0.25) -> Optional[Skill]:
    """Best skill for ``message`` when its score reaches ``threshold``."""
    best = max(skills, key=lambda s: s.score(message), default=None)
    return best if best and best.score(message) >= threshold else None

---
name: account-health-review
description: Use when the user asks how healthy, risky or satisfied a customer account is, or whether it may churn.
tags:
  - crm
---

# Account health review

1. Look up the account with `get_account`.
2. Call `compute_health_score` and `list_open_tickets`.
3. Call `get_usage_trend` and report the three-month trend.
4. Finish with one verdict (healthy, needs attention, at risk) and one recommended next action.