tools.py

"""Hermes tools for nostrsocial — file-backed enclave persisted across calls."""

from __future__ import annotations

import os
from pathlib import Path
from typing import Any, Optional

from nostrsocial import FileStorage, SocialEnclave, Tier

from tools.registry import tool_error, tool_result

_enclave: Optional[SocialEnclave] = None


def _hermes_home() -> Path:
    return Path(os.environ.get("HERMES_HOME", str(Path.home() / ".hermes")))


def _default_storage_path() -> Path:
    return _hermes_home() / ".nostrsocial" / "graph.json"


def _require_enclave() -> SocialEnclave:
    if _enclave is None:
        raise RuntimeError("Social graph not initialized. Call social_init first.")
    return _enclave


def _tier_from(s: str) -> Tier:
    return Tier(s.lower())


# -----------------------------
# social_init
# -----------------------------

SOCIAL_INIT_SCHEMA = {
    "type": "function",
    "function": {
        "name": "social_init",
        "description": (
            "Create or load the social graph enclave. Persisted to "
            "$HERMES_HOME/.nostrsocial/graph.json by default. ⚠ The first "
            "create generates a device secret — losing it makes proxy "
            "identities unrecoverable."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "filepath": {"type": "string", "description": "Override storage path."},
                "force_create": {"type": "boolean", "default": False, "description": "Force a new enclave even if one exists at filepath."},
            },
            "required": [],
        },
    },
}


def handle_social_init(args: dict[str, Any], **kw) -> str:
    global _enclave
    try:
        path = Path(args.get("filepath") or _default_storage_path())
        path.parent.mkdir(parents=True, exist_ok=True)
        storage = FileStorage(str(path))
        force = bool(args.get("force_create", False))
        loaded = None
        if not force:
            loaded = SocialEnclave.load(storage)
        if loaded is None:
            _enclave = SocialEnclave.create(storage=storage)
            _enclave.save()
            return tool_result({"created": True, "path": str(path), "device_secret_export": _enclave.export_secret()})
        _enclave = loaded
        return tool_result({"loaded": True, "path": str(path)})
    except Exception as e:
        return tool_error(f"social_init failed: {type(e).__name__}: {e}")


# -----------------------------
# social_add
# -----------------------------

SOCIAL_ADD_SCHEMA = {
    "type": "function",
    "function": {
        "name": "social_add",
        "description": "Add a contact to the friends list. Tier is one of intimate/close/familiar/known.",
        "parameters": {
            "type": "object",
            "properties": {
                "identifier":   {"type": "string", "description": "Channel-specific id, e.g. an email, npub, handle."},
                "channel":      {"type": "string", "description": "Channel name, e.g. 'nostr', 'email', 'matrix'."},
                "tier":         {"type": "string", "enum": ["intimate", "close", "familiar", "known"]},
                "display_name": {"type": "string"},
                "notes":        {"type": "string"},
                "claimed_npub": {"type": "string", "description": "If contact claims a Nostr npub, store it for later verification."},
            },
            "required": ["identifier", "channel", "tier"],
        },
    },
}


def handle_social_add(args: dict[str, Any], **kw) -> str:
    try:
        enclave = _require_enclave()
        contact = enclave.add(
            identifier=args["identifier"],
            channel=args["channel"],
            tier=_tier_from(args["tier"]),
            display_name=args.get("display_name"),
            notes=args.get("notes"),
            claimed_npub=args.get("claimed_npub"),
        )
        enclave.save()
        return tool_result({
            "proxy_npub": contact.proxy_npub,
            "tier": contact.tier.value if contact.tier else None,
            "display_name": contact.display_name,
        })
    except Exception as e:
        return tool_error(f"social_add failed: {type(e).__name__}: {e}")


# -----------------------------
# social_list
# -----------------------------

SOCIAL_LIST_SCHEMA = {
    "type": "function",
    "function": {
        "name": "social_list",
        "description": "List contacts in the friends list, optionally filtered by tier.",
        "parameters": {
            "type": "object",
            "properties": {
                "tier": {"type": "string", "enum": ["intimate", "close", "familiar", "known"]},
            },
            "required": [],
        },
    },
}


def handle_social_list(args: dict[str, Any], **kw) -> str:
    try:
        enclave = _require_enclave()
        tier = _tier_from(args["tier"]) if args.get("tier") else None
        contacts = enclave._contacts.list_friends(tier=tier)
        out = [
            {"proxy_npub": c.proxy_npub, "display_name": c.display_name,
             "tier": c.tier.value if c.tier else None,
             "channel": c.channel, "interaction_count": c.interaction_count}
            for c in contacts
        ]
        return tool_result({"count": len(out), "contacts": out})
    except Exception as e:
        return tool_error(f"social_list failed: {type(e).__name__}: {e}")


# -----------------------------
# social_promote
# -----------------------------

SOCIAL_PROMOTE_SCHEMA = {
    "type": "function",
    "function": {
        "name": "social_promote",
        "description": "Move a contact to a different trust tier.",
        "parameters": {
            "type": "object",
            "properties": {
                "identifier": {"type": "string"},
                "channel":    {"type": "string"},
                "new_tier":   {"type": "string", "enum": ["intimate", "close", "familiar", "known"]},
            },
            "required": ["identifier", "channel", "new_tier"],
        },
    },
}


def handle_social_promote(args: dict[str, Any], **kw) -> str:
    try:
        enclave = _require_enclave()
        contact = enclave.promote(
            identifier=args["identifier"],
            channel=args["channel"],
            new_tier=_tier_from(args["new_tier"]),
        )
        enclave.save()
        return tool_result({
            "proxy_npub": contact.proxy_npub,
            "new_tier": contact.tier.value if contact.tier else None,
        })
    except Exception as e:
        return tool_error(f"social_promote failed: {type(e).__name__}: {e}")


# -----------------------------
# social_screen
# -----------------------------

SOCIAL_SCREEN_SCHEMA = {
    "type": "function",
    "function": {
        "name": "social_screen",
        "description": "Run text through the guardrails screen — flags hostile/manipulative content.",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "Text to screen."},
            },
            "required": ["text"],
        },
    },
}


def handle_social_screen(args: dict[str, Any], **kw) -> str:
    try:
        enclave = _require_enclave()
        result = enclave.screen(args["text"])
        return tool_result({
            "passes": result.passes,
            "categories": list(result.categories),
            "matched_words": list(result.matched_words),
        })
    except Exception as e:
        return tool_error(f"social_screen failed: {type(e).__name__}: {e}")