naldo_opencode.py

"""Naldo_Bot brain constants and offline-safe helpers for the OpenCode path.

Only Naldo_Bot uses OpenCode. Gemini_Bot keeps its own proxy and must not
import this module's model list as a fallback.

Design notes (do not drift without updating the tests):
- Provider is always ``opencode``. Only FREE opencode models are allowed.
  Never Granite, Gemini, external OpenAI, Anthropic, or any other family.
- No OPENCODE_API_KEY is required: the free opencode models work with cost 0
  in a clean HOME. This module never reads, copies, or logs credential files.
- ``opencode serve`` binds 127.0.0.1 only, on a dedicated internal port.
- The bridge (naldo_opencode_bridge.py) binds 127.0.0.1 only on its own port
  and translates OpenAI-compatible chat calls into OpenCode session calls:
  POST /session -> POST /session/{id}/message, health via GET /global/health.
- OpenCode session message API exposes no temperature field, so the bridge
  cannot forward Voyager's temperature. The requested value is recorded for
  observability; the effective sampling is the provider default.
- OpenCode here is a TEXT GENERATOR for Voyager agents (Curriculum/Action/
  Critic/SkillManager). In OpenCode 1.18.33, message-level ``tools: {}``
  disables tool execution while remaining compatible with the Zen free tier.
  Custom deny-all session permissions are intentionally not sent because Zen
  rejects those customized sessions with FreeTierError/403.
"""

from __future__ import annotations

import json
import os
import subprocess
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

OPENCODE_PROVIDER = "opencode"
OPENCODE_VERSION = "1.18.33"
OPENCODE_NPM_SPEC = "opencode-ai@1.18.33"

PRIMARY_MODEL = "muse-spark-1.3-contributor-free"
FALLBACK_MODELS: tuple[str, ...] = (
    "muse-spark-1.3-contributor-free",
    "fledge-alpha-free",
    "nemotron-3-ultra-free",
)
ALLOWED_MODELS = frozenset(FALLBACK_MODELS)

BRAIN_PROFILE = "naldo-opencode"

RUNTIME_HOME = Path(os.environ.get("VOYAGER_RUNTIME_HOME", "/tmp/voyager-runtime"))
OPENCODE_INSTALL_DIR = Path(os.environ.get("OPENCODE_INSTALL_DIR", str(RUNTIME_HOME / "opencode-cli")))
OPENCODE_HOME_DIR = Path(os.environ.get("OPENCODE_HOME_DIR", str(RUNTIME_HOME / "opencode-home")))
OPENCODE_SESSION_DIR = Path(os.environ.get("OPENCODE_SESSION_DIR", str(RUNTIME_HOME / "opencode-sessions")))

OPENCODE_PORT = int(os.environ.get("OPENCODE_PORT", "7865"))
BRIDGE_PORT = int(os.environ.get("NALDO_BRIDGE_PORT", "7864"))
OPENCODE_BASE = os.environ.get("OPENCODE_BASE", f"http://127.0.0.1:{OPENCODE_PORT}").rstrip("/")
BRIDGE_BASE = os.environ.get("NALDO_BRIDGE_BASE", f"http://127.0.0.1:{BRIDGE_PORT}/v1").rstrip("/")

# Local embeddings stay on the public 7860 endpoint so SkillManager/QA keep
# working. Only generative chat moves to the OpenCode bridge.
LOCAL_EMBED_BASE = os.environ.get("NALDO_EMBED_BASE", "http://127.0.0.1:7860/v1").rstrip("/")

FORBIDDEN_MODEL_SUBSTRINGS = (
    "granite",
    "gemini",
    "gpt-",
    "openai",
    "anthropic",
    "claude",
    "qwen",
)

LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})

TEMPERATURE_NOTE = (
    "OpenCode session message API exposes no temperature field; "
    "Voyager's requested temperature is recorded but not forwarded. "
    "Effective sampling is the opencode provider default."
)


def canonical_model_id(name: str) -> str:
    """Strip an optional ``opencode/`` prefix to the bare free-model id."""
    text = str(name or "").strip()
    if "/" in text:
        provider, _, model = text.partition("/")
        if provider.strip().lower() == OPENCODE_PROVIDER:
            return model.strip()
        return text
    return text


def is_allowed_model(name: str) -> bool:
    """True only for FREE models of the ``opencode`` provider."""
    text = str(name or "").strip()
    if not text:
        return False
    lowered = text.lower()
    for forbidden in FORBIDDEN_MODEL_SUBSTRINGS:
        if forbidden in lowered:
            return False
    if "/" in text:
        provider, _, model = text.partition("/")
        if provider.strip().lower() != OPENCODE_PROVIDER:
            return False
        return model.strip() in ALLOWED_MODELS
    return text in ALLOWED_MODELS


def is_loopback_url(url: str) -> bool:
    try:
        host = (urllib.parse.urlparse(url).hostname or "").lower()
    except Exception:
        return False
    return host in LOOPBACK_HOSTS


def _content(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        parts: list[str] = []
        for item in value:
            if isinstance(item, dict):
                text = item.get("text")
                if isinstance(text, str):
                    parts.append(text)
            elif isinstance(item, str):
                parts.append(item)
        return "\n".join(parts)
    return str(value)


def build_opencode_prompt(messages: list[dict[str, Any]]) -> dict[str, str]:
    """Render OpenAI-style messages into a deterministic role-tagged prompt.

    System messages go to the OpenCode ``system`` field AND stay tagged in the
    prompt body so role boundaries survive a single-text-part transport.
    tool_calls/tool outputs are rendered as inert text; they are never
    executed. OpenCode tools stay disabled server-side.
    """
    systems: list[str] = []
    rendered: list[str] = []
    has_tool_traffic = False
    for raw in messages or []:
        if not isinstance(raw, dict):
            continue
        role = str(raw.get("role", "user")).upper()
        body = _content(raw.get("content"))
        extra = ""
        tool_calls = raw.get("tool_calls")
        if isinstance(tool_calls, list) and tool_calls:
            has_tool_traffic = True
            try:
                extra = "\n[TOOL_CALLS]\n" + json.dumps(tool_calls, ensure_ascii=False)[:4000]
            except Exception:
                extra = "\n[TOOL_CALLS]\n<unserializable>"
        if raw.get("tool_call_id") or role == "TOOL":
            has_tool_traffic = True
        if role == "SYSTEM":
            if body:
                systems.append(body)
            rendered.append(f"[SYSTEM]\n{body}{extra}")
        else:
            rendered.append(f"[{role}]\n{body}{extra}")
    header = (
        "You are the language model inside the Voyager Minecraft agent. "
        "Follow the supplied system and user instructions exactly. "
        "Return only the response format requested by the final prompt; no wrapper commentary. "
        "OpenCode tools are disabled; you are a text generator, not a file/shell executor."
    )
    if has_tool_traffic:
        header += (
            " Note: any TOOL_CALLS/TOOL sections above are rendered as inert text "
            "from the client; do not attempt tool calls, return plain text only."
        )
    system_text = "\n\n".join(systems)
    prompt = header + "\n\n" + "\n\n".join(rendered) if rendered else header
    return {"system": system_text, "prompt": prompt}


def extract_assistant_text(payload: Any) -> str:
    """Return only assistant ``text`` parts from an OpenCode message result."""
    parts: Any = None
    if isinstance(payload, dict):
        parts = payload.get("parts", payload.get("message", {}).get("parts") if isinstance(payload.get("message"), dict) else None)
        if parts is None and isinstance(payload.get("info"), dict):
            # Some shapes nest under info/parts; fall through to generic scan.
            parts = payload.get("parts")
    elif isinstance(payload, list):
        parts = payload
    if not isinstance(parts, list):
        return ""
    texts: list[str] = []
    for part in parts:
        if not isinstance(part, dict):
            continue
        if part.get("type") != "text":
            continue
        text = part.get("text")
        if isinstance(text, str) and text:
            texts.append(text)
    return "".join(texts)


def opencode_message_body(model_id: str, system: str, prompt: str) -> dict[str, Any]:
    """Build POST /session/{id}/message body with tools blocked.

    No temperature is sent: the session API has no such field (see
    TEMPERATURE_NOTE). ``tools: {}`` disables file/shell execution.
    """
    body: dict[str, Any] = {
        "model": {"providerID": OPENCODE_PROVIDER, "modelID": canonical_model_id(model_id)},
        "tools": {},
        "parts": [{"type": "text", "text": prompt}],
    }
    if system:
        body["system"] = system
    return body


def opencode_binary_path(install_dir: Path = OPENCODE_INSTALL_DIR) -> Path:
    name = "opencode.cmd" if os.name == "nt" else "opencode"
    return install_dir / "node_modules" / ".bin" / name


def opencode_serve_command(binary: str | Path, port: int = OPENCODE_PORT) -> list[str]:
    return [str(binary), "serve", "--hostname", "127.0.0.1", "--port", str(port)]


def bridge_command(port: int = BRIDGE_PORT) -> list[str]:
    import sys

    entrypoint = Path(__file__).resolve().parent / "naldo_opencode_bridge.py"
    return [sys.executable, "-u", str(entrypoint)]


def _http_json(
    url: str,
    method: str = "GET",
    body: dict[str, Any] | None = None,
    timeout: float = 10.0,
) -> Any:
    data = json.dumps(body).encode("utf-8") if body is not None else None
    request = urllib.request.Request(url, data=data, method=method)
    if data is not None:
        request.add_header("Content-Type", "application/json")
    with urllib.request.urlopen(request, timeout=timeout) as response:
        raw = response.read().decode("utf-8", errors="replace")
    try:
        return json.loads(raw) if raw else {}
    except ValueError:
        return {"raw": raw}


def check_opencode_health(base: str = OPENCODE_BASE, timeout: float = 5.0) -> dict[str, Any]:
    try:
        payload = _http_json(base.rstrip("/") + "/global/health", timeout=timeout)
        ready = bool(isinstance(payload, dict) and payload.get("healthy") is True)
        return {
            "ok": ready,
            "ready": ready,
            "detail": payload if isinstance(payload, dict) else {"raw": str(payload)[:200]},
        }
    except Exception as exc:
        return {"ok": False, "ready": False, "error": f"{type(exc).__name__}: {str(exc)[:160]}"}


def check_bridge_health(base: str = BRIDGE_BASE, timeout: float = 5.0) -> dict[str, Any]:
    try:
        root = base.rstrip("/")
        health_url = root + "/health" if root.endswith("/v1") else root + "/health"
        # BRIDGE_BASE ends with /v1; health lives at /health (no /v1 prefix).
        if health_url.endswith("/v1/health"):
            health_url = health_url[: -len("/v1/health")] + "/health"
        payload = _http_json(health_url, timeout=timeout)
        ready = bool(isinstance(payload, dict) and payload.get("ready"))
        return {"ok": True, "ready": ready, "detail": payload if isinstance(payload, dict) else {}}
    except Exception as exc:
        return {"ok": False, "ready": False, "error": f"{type(exc).__name__}: {str(exc)[:160]}"}


def opencode_subprocess_env(
    base_env: dict[str, str] | None = None,
    home_dir: Path = OPENCODE_HOME_DIR,
) -> dict[str, str]:
    """Build an isolated child environment without inheriting OpenCode auth."""
    env = dict(base_env if base_env is not None else os.environ)
    config_dir = home_dir / "config"
    data_dir = home_dir / "data"
    for directory in (home_dir, config_dir, data_dir, OPENCODE_SESSION_DIR):
        directory.mkdir(parents=True, exist_ok=True)
    env.pop("OPENCODE_API_KEY", None)
    env["HOME"] = str(home_dir)
    env["XDG_CONFIG_HOME"] = str(config_dir)
    env["XDG_DATA_HOME"] = str(data_dir)
    return env


def _probe_opencode(binary: Path, timeout: int) -> str:
    try:
        proc = subprocess.run(
            [str(binary), "--version"],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            timeout=timeout,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        return ""
    return (proc.stdout or "").strip() if proc.returncode == 0 else ""


def ensure_opencode_installed(
    install_dir: Path = OPENCODE_INSTALL_DIR,
    timeout: int = 60,
) -> tuple[str, str]:
    """Install the pinned CLI locally; never require a global npm prefix/root."""
    install_dir.mkdir(parents=True, exist_ok=True)
    binary = opencode_binary_path(install_dir)
    reported = _probe_opencode(binary, timeout)
    if reported == OPENCODE_VERSION:
        return str(binary), reported

    npm = "npm.cmd" if os.name == "nt" else "npm"
    proc = subprocess.run(
        [
            npm, "install", "--no-audit", "--no-fund",
            "--prefix", str(install_dir), OPENCODE_NPM_SPEC,
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        timeout=600,
        check=False,
    )
    output = proc.stdout or ""
    if proc.returncode != 0:
        raise RuntimeError(f"opencode local install failed rc={proc.returncode}: {output[-2000:]}")
    reported = _probe_opencode(binary, timeout)
    if reported != OPENCODE_VERSION:
        raise RuntimeError(
            f"opencode CLI version mismatch after local install: expected={OPENCODE_VERSION!r} got={reported!r}"
        )
    return str(binary), reported