| """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_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): |
| |
| 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" |
| |
| 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 |
|
|
