opencode_config.py

"""OpenCode configuration for remote, keyless Minecraft supervision."""
from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any

OPENCODE_VERSION = "1.18.34"
MINE_BASE = "https://erinaldordso-mine.hf.space/v1"
MAIN_MODEL = "qwen3-coder-30b-a3b-instruct"
AUX_MODEL = "granite-4.0-h-tiny"
# Coach-only model: high-level supervision runs SOLELY on the free OpenCode
# Muse model. No Mine provider is configured here; other bots/Spaces that use
# the Mine Space keep their own wiring and are unaffected.
MUSE_MODEL_ID = "muse-spark-1.3-contributor-free"
MUSE_MODEL = "opencode/" + MUSE_MODEL_ID
COACH_PROMPT = """You supervise the upstream AltoClef Minecraft survival speedrun bot.
AltoClef performs real movement, gear progression, stronghold search, portal
activation and dragon combat on difficulty Normal. You are a text decider, not
a file/shell executor: use no tools, run no commands, edit nothing. Observe
only the supplied telemetry and reply ONLY with one compact JSON object:
{"action":"continue","reason":"brief factual explanation"}.
Allowed actions: start, continue, resume. Start/resume runs AltoClef @gamer only
when its task is inactive. Prefer continue while a task is active. Never invent
world observations or success. Naldo plays Survival with a retained pre-existing
inventory (assisted run). Credits observed after entering the End are required
to claim completion. Keep reason under 25 words.
"""


def configuration() -> dict[str, Any]:
    # Supported Zen free-tier shape (probed 2026-10-03, opencode 1.18.33,
    # keyless isolated HOME, empty stdin): NO custom agent at all. Upstream
    # request.ts uses `agent.prompt OR SystemPrompt.provider(model)`, and Zen
    # rejects the tested override shapes with
    # `FreeTierError: ... can only be used from within OpenCode` (403):
    # prompt override, tools lockdown, blanket permission deny, and even a
    # read-only permission shape (*-deny + read/glob/grep-allow +
    # external_directory-deny) each reproduce it. Bare description+mode and the
    # built-ins (build, plan) succeed. The built-in `plan` agent is used:
    # official Muse system prompt intact, read-only by design
    # ("Plan mode. Disallows all edit tools"). All coach instructions live at
    # user level (COACH_PROMPT + telemetry in the run message). There is
    # intentionally no Mine provider either, so nothing falls back to Mine.
    return {
        "$schema": "https://opencode.ai/config.json",
        "enabled_providers": ["opencode"],
        "model": MUSE_MODEL,
        "small_model": MUSE_MODEL,
        "share": "disabled",
        "autoupdate": False,
        "default_agent": "plan",
    }


def coach_user_message(telemetry: dict[str, Any]) -> str:
    """Build the user-level run message: instructions + real telemetry."""
    compact = {key: telemetry.get(key) for key in [
        "connected", "dimension", "position", "creative", "task_active",
        "tasks", "entered_end", "credits_observed", "error"]}
    return COACH_PROMPT + "\nReal telemetry: " + json.dumps(compact)


def child_environment(root: Path) -> dict[str, str]:
    # Independent configuration/data prevents inheriting other CLI credentials
    # or reading the Xbox relay directory. Models are reached without a key.
    env = dict(os.environ)
    for name in list(env):
        if name.endswith("API_KEY") or name.startswith("GEMINI_BOT_KEY_") or name in {
            "HF_TOKEN", "HUGGING_FACE_HUB_TOKEN", "VOYAGER_API_KEY", "GITHUB_TOKEN", "GH_TOKEN"
        }:
            env.pop(name, None)
    env.update({
        "XDG_CONFIG_HOME": str(root / "config"),
        "XDG_DATA_HOME": str(root / "data"),
        "XDG_CACHE_HOME": str(root / "cache"),
        "OPENCODE_CONFIG": str(root / "opencode.json"),
        "OPENCODE_DISABLE_AUTOUPDATE": "true",
        "OPENCODE_DISABLE_DEFAULT_PLUGINS": "true",
        "OPENCODE_DISABLE_SHARE": "true",
    })
    return env


def parse_decision(output: str) -> dict[str, str]:
    texts: list[str] = []
    for line in output.splitlines():
        try:
            event = json.loads(line)
        except (ValueError, TypeError):
            continue
        if event.get("type") == "error":
            raise RuntimeError(str(event.get("error") or event)[:300])
        part = event.get("part") or {}
        if event.get("type") == "text" and part.get("text"):
            texts.append(part["text"])
    raw = "\n".join(texts).strip()
    if raw.startswith("```"):
        raw = "\n".join(raw.splitlines()[1:-1]).strip()
    try:
        decision = json.loads(raw)
    except ValueError as exc:
        raise RuntimeError("OpenCode did not return a valid decision: " + raw[:200]) from exc
    if not isinstance(decision, dict) or decision.get("action") not in {"start", "continue", "resume"}:
        raise RuntimeError("Unsupported OpenCode decision")
    return {"action": decision["action"], "reason": str(decision.get("reason") or "")[:240]}


def goal_completed(observation: dict[str, Any]) -> bool:
    # A disappearing/unloaded dragon or an inactive task never means victory.
    return bool(observation.get("entered_end") and observation.get("credits_observed"))