Elydora-Python-SDK/elydora/plugins/opencode.py

"""OpenCode integration using its current ESM plugin API."""

from __future__ import annotations

import glob
import json
import os
import sys

from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script


PLUGIN_DIR = os.path.join(os.path.expanduser("~"), ".config", "opencode", "plugins")
PLUGIN_FILENAME = "elydora-audit.mjs"
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")


class OpenCodePlugin(AgentPlugin):
    """Install the Elydora guard and audit hooks for OpenCode."""

    def _plugin_path(self) -> str:
        return os.path.join(PLUGIN_DIR, PLUGIN_FILENAME)

    @staticmethod
    def _hook_path(agent_id: str) -> str:
        return os.path.join(ELYDORA_DIR, agent_id, "hook.py")

    def install(self, config: InstallConfig) -> None:
        agent_id = config.get("agent_id", "")
        guard_script_path = config.get("guard_script_path", "")
        if not agent_id:
            raise ValueError("agent_id is required")
        if not guard_script_path:
            raise ValueError("guard_script_path is required")

        agent_dir = os.path.join(ELYDORA_DIR, agent_id)
        os.makedirs(agent_dir, exist_ok=True)
        config_data = {
            "org_id": config.get("org_id", ""),
            "agent_id": agent_id,
            "kid": config.get("kid", ""),
            "base_url": config.get("base_url", "https://api.elydora.com"),
            "token": config.get("token", ""),
            "agent_name": config.get("agent_name", "opencode"),
        }
        _write_text(
            os.path.join(agent_dir, "config.json"),
            json.dumps(config_data, indent=2) + "\n",
            0o600,
        )
        _write_text(
            os.path.join(agent_dir, "private.key"),
            config.get("private_key", ""),
            0o600,
        )

        hook_path = self._hook_path(agent_id)
        hook_script = generate_hook_script(
            org_id=config.get("org_id", ""),
            agent_id=agent_id,
            private_key=config.get("private_key", ""),
            kid=config.get("kid", ""),
            base_url=config.get("base_url", "https://api.elydora.com"),
        )
        _write_text(hook_path, hook_script, 0o700)

        os.makedirs(PLUGIN_DIR, exist_ok=True)
        _write_text(
            self._plugin_path(),
            _generate_js_plugin(
                hook_script_path=hook_path,
                guard_script_path=guard_script_path,
            ),
            0o700,
        )
        print("Elydora hook installed for OpenCode.")
        print(f"  Plugin: {self._plugin_path()}")

    def uninstall(self, agent_id: str = "") -> None:
        try:
            os.remove(self._plugin_path())
        except FileNotFoundError:
            pass
        print("Elydora hook uninstalled from OpenCode.")

    def status(self) -> PluginStatus:
        plugin_path = self._plugin_path()
        runtime_ready = False
        for config_path in glob.glob(os.path.join(ELYDORA_DIR, "*", "config.json")):
            try:
                with open(config_path, "r", encoding="utf-8") as file:
                    config = json.load(file)
            except (OSError, json.JSONDecodeError):
                continue
            agent_dir = os.path.dirname(config_path)
            if config.get("agent_name") == "opencode":
                runtime_ready = os.path.isfile(os.path.join(agent_dir, "hook.py")) and os.path.isfile(
                    os.path.join(agent_dir, "guard.py")
                )
                if runtime_ready:
                    break

        installed = os.path.isfile(plugin_path) and runtime_ready
        details = f"Plugin: {plugin_path}" if installed else "Not installed"
        return PluginStatus(installed=installed, agent="opencode", details=details)


def _write_text(path: str, content: str, mode: int) -> None:
    with open(path, "w", encoding="utf-8") as file:
        file.write(content)
    try:
        os.chmod(path, mode)
    except OSError:
        pass


def _generate_js_plugin(*, hook_script_path: str, guard_script_path: str) -> str:
    return f'''// Elydora Audit Plugin for OpenCode
// Generated by elydora-sdk — DO NOT EDIT

import {{ spawn, spawnSync }} from 'node:child_process';

const PYTHON_EXE = {_js_str(sys.executable)};
const HOOK_SCRIPT_PATH = {_js_str(hook_script_path)};
const GUARD_SCRIPT_PATH = {_js_str(guard_script_path)};

export const ElydoraAuditPlugin = async () => ({{
  "tool.execute.before": async () => {{
    const result = spawnSync(PYTHON_EXE, [GUARD_SCRIPT_PATH], {{
      timeout: 5000,
      stdio: ['pipe', 'ignore', 'pipe'],
    }});
    if (result.error) {{
      throw new Error('Elydora guard failed: ' + result.error.message);
    }}
    if (result.status !== 0) {{
      const message = result.stderr?.toString().trim() || 'Elydora guard blocked tool execution.';
      throw new Error(message);
    }}
  }},

  "tool.execute.after": async (input) => {{
    const event = JSON.stringify({{
      tool_name: input?.tool || 'unknown',
      tool_input: input?.args || {{}},
      session_id: input?.sessionID || 'unknown',
    }});
    const child = spawn(PYTHON_EXE, [HOOK_SCRIPT_PATH], {{
      detached: true,
      stdio: ['pipe', 'ignore', 'ignore'],
    }});
    child.on('error', (error) => {{
      console.error('[elydora] audit hook failed:', error.message);
    }});
    child.stdin.end(event);
    child.unref();
  }},
}});
'''


def _js_str(value: str) -> str:
    return json.dumps(value)