hermes-identity-plugin/__init__.py

"""
hermes-identity-plugin (identity)

Resolves platform-specific user IDs and kanban workers to stable Honcho
peer names. The plugin provides hooks + slash command. The actual peer name
injection into Honcho requires a ~6-line change in the bundled Honcho plugin
(plugins/memory/honcho/session.py) — see docs/honcho-injector.md.

Config: /opt/data/identity-config.json (persistent across container rebuilds)
"""

from __future__ import annotations

import json
import logging
import os
import re
import sqlite3
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)

CONFIG_PATH = Path("/opt/data/identity-config.json")

# ── Config ──────────────────────────────────────────────────────────────────


def _default_config() -> dict[str, Any]:
    return {
        "mappings": [],
        "boards": {},
        "fallback_peer": "user",
        "enforce_context_peer": True,
    }


def load_config() -> dict[str, Any]:
    if not CONFIG_PATH.exists():
        return _default_config()
    try:
        return json.loads(CONFIG_PATH.read_text())
    except (json.JSONDecodeError, OSError) as exc:
        logger.warning("identity: config parse error: %s", exc)
        return _default_config()


def save_config(cfg: dict[str, Any]) -> None:
    CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
    CONFIG_PATH.write_text(json.dumps(cfg, indent=2))
    logger.info("identity: config saved to %s", CONFIG_PATH)


# ── Resolution ──────────────────────────────────────────────────────────────


def resolve_peer(platform: str, platform_id: str) -> str | None:
    """Resolve a platform+ID to a canonical peer name."""
    cfg = load_config()
    for m in cfg.get("mappings", []):
        if m.get("platform") == platform and str(m.get("id", "")) == str(platform_id):
            return m["peer"]
    return cfg.get("fallback_peer")


def resolve_board_peer(board_slug: str | None) -> str | None:
    """Resolve a kanban board slug to a peer name."""
    if not board_slug:
        return None
    cfg = load_config()
    return cfg.get("boards", {}).get(board_slug)


def get_peer_name() -> str | None:
    """Resolve peer name for the current process context.

    Priority:
    1. HERMES_HONCHO_PEER_NAME env var (set by kanban dispatcher or manually)
    2. Kanban worker: task body context_peer → board config → fallback
    """
    explicit = os.environ.get("HERMES_HONCHO_PEER_NAME")
    if explicit:
        return explicit

    task_id = os.environ.get("HERMES_KANBAN_TASK")
    board = os.environ.get("HERMES_KANBAN_BOARD")
    db_path = os.environ.get("HERMES_KANBAN_DB")
    if task_id and db_path:
        return _resolve_kanban_peer(task_id, board, db_path)

    return None


def _resolve_kanban_peer(task_id: str, board: str | None, db_path: str) -> str | None:
    """Read kanban task body for context_peer, fallback to board config."""
    try:
        db = Path(db_path)
        if not db.exists():
            return None
        conn = sqlite3.connect(str(db))
        conn.row_factory = sqlite3.Row
        try:
            row = conn.execute(
                "SELECT body FROM tasks WHERE id = ?", (task_id,)
            ).fetchone()
            if not row:
                return None
            body = row["body"] or ""
            peer = _extract_context_peer(body)
            if peer:
                return peer
            return resolve_board_peer(board)
        finally:
            conn.close()
    except Exception as exc:
        logger.debug("identity: kanban peer resolution error: %s", exc)
        return None


_CONTEXT_PEER_RE = re.compile(r"```metadata\s*\ncontext_peer:\s*(\S+)", re.MULTILINE)
_INLINE_PEER_RE = re.compile(r"context_peer:\s*(\S+)")


def _extract_context_peer(body: str) -> str | None:
    m = _CONTEXT_PEER_RE.search(body)
    if m:
        return m.group(1)
    m = _INLINE_PEER_RE.search(body)
    if m:
        return m.group(1)
    return None


# ── Enforce context_peer on kanban task creation ────────────────────────────
# Profiles (Claire, Ashley, etc.) MUST include ``context_peer: <name>`` in
# every kanban task body. The kanban-worker skill references this convention.
# The Honcho injector (see docs/honcho-injector.md) reads it at worker start.


def enforce_context_peer(body: str | None) -> str | None:
    """Return error message if body lacks context_peer, None if OK."""
    cfg = load_config()
    if not cfg.get("enforce_context_peer", True):
        return None
    if not body or "context_peer:" not in body:
        return (
            "Missing required `context_peer: <name>` in task body. "
            "Add a metadata block:\n"
            "```metadata\ncontext_peer: thierry\n```\n"
            "Replace `thierry` with the intended user's peer name."
        )
    peer = _extract_context_peer(body)
    if not peer:
        return "context_peer: found in body but could not be parsed. Use format: context_peer: <name>"
    return None


# ── Hooks ───────────────────────────────────────────────────────────────────


def _pre_gateway_dispatch(event: Any, gateway: Any, session_store: Any, **kw) -> dict | None:
    """Log resolved peer identity before dispatch."""
    platform = getattr(event, "platform", "unknown")
    user_id = getattr(event, "user_id", "unknown")
    resolved = resolve_peer(platform, user_id)

    if resolved:
        logger.debug("identity: gateway platform=%s user_id=%s → peer=%s", platform, user_id, resolved)
    else:
        logger.info("identity: unmapped gateway user platform=%s user_id=%s", platform, user_id)
    return None  # allow normal dispatch


def _on_session_start(**kw: Any) -> None:
    """Log resolved identity at session start."""
    peer = get_peer_name()
    task_id = os.environ.get("HERMES_KANBAN_TASK")
    if task_id:
        if peer:
            logger.info("identity: kanban worker task=%s peer=%s", task_id, peer)
        else:
            logger.info("identity: kanban worker task=%s no peer → user-default", task_id)
    elif peer:
        logger.debug("identity: session peer=%s", peer)


# ── Slash command ───────────────────────────────────────────────────────────


def _cmd_identity(raw_args: str) -> str:
    """Handle ``/identity`` slash command."""
    args = raw_args.strip().split()
    if not args:
        return _show_status()

    cmd = args[0]
    if cmd == "status":
        return _show_status()
    elif cmd == "list":
        return _list_mappings()
    elif cmd in ("add", "set") and len(args) >= 4:
        # /identity add discord 479136126737711105 thierry
        platform, pid, peer_name = args[1], args[2], args[3]
        return _add_mapping(platform, pid, peer_name)
    elif cmd == "rm" and len(args) >= 2:
        idx = int(args[1]) if args[1].isdigit() else -1
        return _remove_mapping(idx)
    elif cmd == "board" and len(args) >= 3:
        slug, peer_name = args[1], args[2]
        return _set_board(slug, peer_name)
    elif cmd == "help":
        return _help_text()
    return _help_text()


def _show_status() -> str:
    cfg = load_config()
    lines = [
        "**Identity Plugin**",
        f"Config: `{CONFIG_PATH}`",
        f"Mappings: {len(cfg.get('mappings', []))}",
        f"Boards: {len(cfg.get('boards', {}))}",
        f"Enforce context_peer: {cfg.get('enforce_context_peer', True)}",
        f"Fallback peer: {cfg.get('fallback_peer', 'user')}",
        "",
        "Current context:",
    ]
    peer = get_peer_name()
    if peer:
        lines.append(f"  Resolved peer: **{peer}**")
    task = os.environ.get("HERMES_KANBAN_TASK")
    if task:
        lines.append(f"  Kanban task: `{task}`")
    board = os.environ.get("HERMES_KANBAN_BOARD")
    if board:
        lines.append(f"  Board: `{board}`")
    return "\n".join(lines)


def _list_mappings() -> str:
    cfg = load_config()
    mappings = cfg.get("mappings", [])
    if not mappings:
        return "No mappings configured."
    lines = ["**Identity mappings:**", ""]
    for i, m in enumerate(mappings):
        lines.append(f"  [{i}] `{m.get('platform')}` `{m.get('id')}` → **{m.get('peer')}**")
    boards = cfg.get("boards", {})
    if boards:
        lines.append("")
        lines.append("**Board peers:**")
        for slug, peer in boards.items():
            lines.append(f"  `{slug}` → **{peer}**")
    return "\n".join(lines)


def _add_mapping(platform: str, pid: str, peer_name: str) -> str:
    cfg = load_config()
    cfg.setdefault("mappings", []).append({"platform": platform, "id": pid, "peer": peer_name})
    save_config(cfg)
    return f"Added mapping: `{platform}` `{pid}` → **{peer_name}**"


def _remove_mapping(idx: int) -> str:
    cfg = load_config()
    mappings = cfg.get("mappings", [])
    if idx < 0 or idx >= len(mappings):
        return f"Invalid index {idx}. Use `list` to see indices."
    removed = mappings.pop(idx)
    save_config(cfg)
    return f"Removed: `{removed.get('platform')}` `{removed.get('id')}` → **{removed.get('peer')}**"


def _set_board(slug: str, peer_name: str) -> str:
    cfg = load_config()
    cfg.setdefault("boards", {})[slug] = peer_name
    save_config(cfg)
    return f"Board `{slug}` → **{peer_name}**"


def _help_text() -> str:
    return """**/identity** — manage identity mappings

Commands:
  `status`              Show config status and current context
  `list`                List all mappings and board peers
  `add discord <id> <peer>`  Add a platform→peer mapping
  `rm <index>`          Remove mapping by index
  `board <slug> <peer>` Set a kanban board's default peer
  `help`                Show this help"""


# ── Plugin entry point ──────────────────────────────────────────────────────


def register(ctx: Any) -> None:
    """Register identity plugin hooks and slash command."""
    ctx.register_hook("pre_gateway_dispatch", _pre_gateway_dispatch)
    ctx.register_hook("on_session_start", _on_session_start)
    ctx.register_command(
        name="identity",
        handler=_cmd_identity,
        description="Manage identity peer mappings",
        args_hint="status|list|add|rm|board|help",
    )
    logger.info("identity-plugin: registered hooks (pre_gateway_dispatch, on_session_start) and /identity command")