tools.py

"""Hermes tools for social-alignment — the yellow-line ethical compass.

Pure evaluation: no I/O beyond an in-memory enclave. STOP severity is
code-enforced — alignment_check returns it as a non-negotiable signal."""

from __future__ import annotations

import enum
from dataclasses import asdict, is_dataclass
from typing import Any, Optional

from social_alignment import ActionDomain, AlignmentEnclave

from tools.registry import tool_error, tool_result

_enclave: Optional[AlignmentEnclave] = None


def _require() -> AlignmentEnclave:
    if _enclave is None:
        raise RuntimeError("Alignment enclave not initialized. Call alignment_init first.")
    return _enclave


def _serialize(obj: Any) -> Any:
    if isinstance(obj, enum.Enum):
        return obj.value
    if is_dataclass(obj):
        return _serialize(asdict(obj))
    if isinstance(obj, (list, tuple)):
        return [_serialize(x) for x in obj]
    if isinstance(obj, dict):
        return {k: _serialize(v) for k, v in obj.items()}
    return obj


def _domain(s: str) -> ActionDomain:
    return ActionDomain(s.lower())


# -----------------------------
# alignment_init
# -----------------------------

ALIGNMENT_INIT_SCHEMA = {
    "type": "function",
    "function": {
        "name": "alignment_init",
        "description": "Create the alignment enclave for this session — the agent's compass.",
        "parameters": {
            "type": "object",
            "properties": {
                "owner_npub": {"type": "string"},
                "owner_name": {"type": "string"},
            },
            "required": [],
        },
    },
}


def handle_alignment_init(args: dict[str, Any], **kw) -> str:
    global _enclave
    try:
        _enclave = AlignmentEnclave.create(
            owner_npub=args.get("owner_npub") or "",
            owner_name=args.get("owner_name") or "",
        )
        return tool_result({
            "owner_npub": args.get("owner_npub") or "",
            "owner_name": args.get("owner_name") or "",
            "decision_count": _enclave.decision_count,
        })
    except Exception as e:
        return tool_error(f"alignment_init failed: {type(e).__name__}: {e}")


# -----------------------------
# alignment_check
# -----------------------------

ALIGNMENT_CHECK_SCHEMA = {
    "type": "function",
    "function": {
        "name": "alignment_check",
        "description": (
            "Run the yellow-line five-lens check on a proposed action. Returns "
            "severity (CLEAR / CAUTION / YIELD / STOP). STOP is code-enforced — "
            "if returned, the agent must NOT proceed."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "domain": {
                    "type": "string",
                    "enum": ["sign", "pay", "publish", "send", "schedule", "execute", "disclose", "connect", "modify", "escalate"],
                },
                "description":              {"type": "string"},
                "involves_secrets":         {"type": "boolean", "default": False},
                "involves_money":           {"type": "boolean", "default": False},
                "money_amount_sats":        {"type": "integer", "default": 0},
                "involves_publication":     {"type": "boolean", "default": False},
                "involves_communication":   {"type": "boolean", "default": False},
                "recipient_trust_tier":     {"type": "string", "enum": ["intimate", "close", "familiar", "known", "unknown"]},
                "is_reversible":            {"type": "boolean", "default": True},
                "is_novel":                 {"type": "boolean"},
                "confidence":               {"type": "number", "default": 0.5, "description": "0.0-1.0."},
                "request_origin":           {"type": "string", "default": "self"},
                "resembles_known_attack":   {"type": "boolean", "default": False},
                "crosses_trust_boundary":   {"type": "boolean", "default": False},
            },
            "required": ["domain", "description"],
        },
    },
}


def handle_alignment_check(args: dict[str, Any], **kw) -> str:
    try:
        enclave = _require()
        result = enclave.check(
            domain=_domain(args["domain"]),
            description=args["description"],
            involves_secrets=bool(args.get("involves_secrets", False)),
            involves_money=bool(args.get("involves_money", False)),
            money_amount_sats=int(args.get("money_amount_sats", 0)),
            involves_publication=bool(args.get("involves_publication", False)),
            involves_communication=bool(args.get("involves_communication", False)),
            recipient_trust_tier=args.get("recipient_trust_tier"),
            is_reversible=bool(args.get("is_reversible", True)),
            is_novel=args.get("is_novel"),
            confidence=float(args.get("confidence", 0.5)),
            request_origin=args.get("request_origin") or "self",
            resembles_known_attack=bool(args.get("resembles_known_attack", False)),
            crosses_trust_boundary=bool(args.get("crosses_trust_boundary", False)),
        )
        return tool_result(_serialize(result))
    except Exception as e:
        return tool_error(f"alignment_check failed: {type(e).__name__}: {e}")


# -----------------------------
# alignment_proceed
# -----------------------------

ALIGNMENT_PROCEED_SCHEMA = {
    "type": "function",
    "function": {
        "name": "alignment_proceed",
        "description": "Record that the agent proceeded with the most recent checked action. Optional owner-override flag.",
        "parameters": {
            "type": "object",
            "properties": {
                "owner_overrode":  {"type": "boolean", "default": False},
                "owner_feedback":  {"type": "string"},
            },
            "required": [],
        },
    },
}


def handle_alignment_proceed(args: dict[str, Any], **kw) -> str:
    try:
        decision = _require().record_proceeded(
            owner_overrode=bool(args.get("owner_overrode", False)),
            owner_feedback=args.get("owner_feedback") or "",
        )
        return tool_result(_serialize(decision))
    except Exception as e:
        return tool_error(f"alignment_proceed failed: {type(e).__name__}: {e}")


# -----------------------------
# alignment_defer
# -----------------------------

ALIGNMENT_DEFER_SCHEMA = {
    "type": "function",
    "function": {
        "name": "alignment_defer",
        "description": "Record that the agent deferred (escalated) to the human on the most recent checked action.",
        "parameters": {
            "type": "object",
            "properties": {
                "owner_feedback": {"type": "string"},
            },
            "required": [],
        },
    },
}


def handle_alignment_defer(args: dict[str, Any], **kw) -> str:
    try:
        decision = _require().record_deferred(owner_feedback=args.get("owner_feedback") or "")
        return tool_result(_serialize(decision))
    except Exception as e:
        return tool_error(f"alignment_defer failed: {type(e).__name__}: {e}")


# -----------------------------
# alignment_wisdom
# -----------------------------

ALIGNMENT_WISDOM_SCHEMA = {
    "type": "function",
    "function": {
        "name": "alignment_wisdom",
        "description": "Return the wisdom report — track record of decisions, override rates, and learning over a window.",
        "parameters": {
            "type": "object",
            "properties": {
                "window": {"type": "integer", "default": 100, "description": "Number of recent decisions to consider."},
            },
            "required": [],
        },
    },
}


def handle_alignment_wisdom(args: dict[str, Any], **kw) -> str:
    try:
        report = _require().wisdom(window=int(args.get("window", 100)))
        return tool_result(_serialize(report))
    except Exception as e:
        return tool_error(f"alignment_wisdom failed: {type(e).__name__}: {e}")