tests/test_naldo_opencode.py

"""Naldo_Bot on OpenCode only: offline unit tests.

No network, no model, no Minecraft. Pure stdlib plus text-level assertions
on the deployment files.
"""

from __future__ import annotations

import ast
import sys
from pathlib import Path
from types import SimpleNamespace
from unittest import mock

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

import naldo_opencode as brain  # noqa: E402
import naldo_opencode_bridge as bridge_module  # noqa: E402


def test_primary_and_fallbacks_are_all_opencode_free():
    assert brain.OPENCODE_PROVIDER == "opencode"
    assert brain.PRIMARY_MODEL == "muse-spark-1.3-contributor-free"
    assert brain.FALLBACK_MODELS[0] == brain.PRIMARY_MODEL
    assert set(brain.FALLBACK_MODELS) == {
        "muse-spark-1.3-contributor-free",
        "fledge-alpha-free",
        "nemotron-3-ultra-free",
    }
    for model in brain.FALLBACK_MODELS:
        assert brain.is_allowed_model(model)
        assert brain.is_allowed_model(f"opencode/{model}")
        lowered = model.lower()
        for forbidden in ("granite", "gemini", "gpt", "claude", "anthropic", "qwen"):
            assert forbidden not in lowered


def test_no_granite_gemini_fallback_anywhere():
    for name in ("granite", "gemini", "gpt-", "anthropic", "claude"):
        assert name not in {m.lower() for m in brain.FALLBACK_MODELS}
    stack = (ROOT / "voyager_stack.py").read_text(encoding="utf-8")
    assert "granite-4.0-h-tiny" not in stack, "Naldo stack must not reference Granite anymore"
    bridge = (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8")
    lowered = bridge.lower()
    for forbidden in ("granite", "gemini", "anthropic"):
        assert forbidden not in lowered, f"bridge leaked {forbidden}"
    runner = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
    # Gemini_Bot's own lock stays; Naldo's lock must not accept granite.
    assert "naldo-granite" not in runner


def test_brain_lock_requires_opencode_profile_base_and_models():
    runner = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
    assert "NALDO_BRAIN_PROFILE" in runner or "naldo-opencode" in runner
    assert "http://127.0.0.1:7864/v1" in runner
    assert "muse-spark-1.3-contributor-free" in runner
    assert "_naldo_model_allowed" in runner
    # Rejected models: granite/gemini must not pass the predicate.
    assert not brain.is_allowed_model("granite-4.0-h-tiny")
    assert not brain.is_allowed_model("gemini-3.5-flash-lite")
    assert not brain.is_allowed_model("gpt-4o")
    assert brain.is_allowed_model("fledge-alpha-free")
    assert brain.is_allowed_model("nemotron-3-ultra-free")


def test_bridge_formats_messages_without_confusing_roles():
    messages = [
        {"role": "system", "content": "You are a planner."},
        {"role": "user", "content": "Do X"},
        {"role": "assistant", "content": "Plan Y"},
        {"role": "tool", "content": "tool output Z"},
        {"role": "user", "content": "Now?", "tool_calls": [{"id": "1", "function": {"name": "f", "arguments": "{}"}}]},
    ]
    rendered = brain.build_opencode_prompt(messages)
    assert "[SYSTEM]" in rendered["prompt"]
    assert "[USER]" in rendered["prompt"]
    assert "[ASSISTANT]" in rendered["prompt"]
    assert "[TOOL]" in rendered["prompt"]
    assert "You are a planner." in rendered["system"]
    assert "TOOL_CALLS" in rendered["prompt"]
    assert "text generator, not a file/shell executor" in rendered["prompt"]


def test_bridge_extracts_only_text_parts():
    payload = {
        "info": {"role": "assistant"},
        "parts": [
            {"type": "reasoning", "thinking": "secret plan"},
            {"type": "text", "text": "hello "},
            {"type": "tool", "name": "bash", "input": "rm -rf /"},
            {"type": "text", "text": "world"},
        ],
    }
    assert brain.extract_assistant_text(payload) == "hello world"
    assert brain.extract_assistant_text({"parts": []}) == ""
    assert brain.extract_assistant_text({}) == ""


def test_sessions_use_free_tier_compatible_tool_sandbox(tmp_path):
    captured = {}

    def fake_http(url, method="GET", body=None, timeout=0, retries=0):
        captured.update({"url": url, "method": method, "body": body})
        return {"id": "ses_test"}

    with mock.patch.object(bridge_module, "OPENCODE_SESSION_DIR", tmp_path), \
         mock.patch.object(bridge_module, "_http_json", side_effect=fake_http):
        session_id = bridge_module._create_session("muse-spark-1.3-contributor-free")

    assert session_id == "ses_test"
    assert captured["method"] == "POST"
    assert captured["url"].endswith("/session")
    assert captured["body"]["model"] == {
        "id": "muse-spark-1.3-contributor-free",
        "providerID": "opencode",
    }
    assert captured["body"]["location"] == {"directory": str(tmp_path)}
    assert "permission" not in captured["body"], "custom permissions trigger Zen free-tier 403"

    message = brain.opencode_message_body("muse-spark-1.3-contributor-free", "sys", "do X")
    assert message["tools"] == {}, "empty tools map is the verified free-tier tool sandbox"
    assert "temperature" not in message


def test_loopback_only():
    assert brain.is_loopback_url(brain.OPENCODE_BASE)
    assert brain.is_loopback_url(brain.BRIDGE_BASE)
    assert "0.0.0.0" not in (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8")
    stack = (ROOT / "voyager_stack.py").read_text(encoding="utf-8")
    assert "opencode_serve_command(opencode_binary" in stack
    assert 'host="127.0.0.1"' in (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8")
    assert "shell=True" not in (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8")
    assert "shell=True" not in (ROOT / "naldo_opencode.py").read_text(encoding="utf-8")
    serve = brain.opencode_serve_command("/tmp/opencode-local", 7865)
    assert serve[:3] == ["/tmp/opencode-local", "serve", "--hostname"]
    assert "127.0.0.1" in serve
    assert "0.0.0.0" not in serve


def test_no_credential_dependency_or_leak(tmp_path):
    bridge_src = (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8")
    assert "auth.json" not in bridge_src
    assert 'os.environ.get("OPENCODE_API_KEY"' not in bridge_src
    assert 'os.environ["OPENCODE_API_KEY"]' not in bridge_src

    base = {"OPENCODE_API_KEY": "must-not-survive", "PATH": "x"}
    env = brain.opencode_subprocess_env(base, tmp_path / "home")
    assert "OPENCODE_API_KEY" not in env
    assert env["PATH"] == "x"
    assert env["HOME"] == str(tmp_path / "home")
    assert env["XDG_CONFIG_HOME"] == str(tmp_path / "home" / "config")
    assert env["XDG_DATA_HOME"] == str(tmp_path / "home" / "data")
    assert base["OPENCODE_API_KEY"] == "must-not-survive", "parent env must not be mutated"
    assert "TEMPERATURE_NOTE" in (ROOT / "naldo_opencode.py").read_text(encoding="utf-8")


def test_local_opencode_install_is_pinned_and_non_root(tmp_path):
    calls = []

    def fake_run(args, **kwargs):
        calls.append(list(args))
        return SimpleNamespace(returncode=0, stdout="installed")

    with mock.patch.object(brain, "_probe_opencode", side_effect=["", brain.OPENCODE_VERSION]), \
         mock.patch.object(brain.subprocess, "run", side_effect=fake_run):
        binary, version = brain.ensure_opencode_installed(tmp_path / "cli")

    assert version == brain.OPENCODE_VERSION
    assert Path(binary) == brain.opencode_binary_path(tmp_path / "cli")
    install = calls[0]
    assert "-g" not in install and "--global" not in install
    assert "--prefix" in install
    assert install[install.index("--prefix") + 1] == str(tmp_path / "cli")
    assert brain.OPENCODE_NPM_SPEC in install


def test_max_tokens_is_telemetry_not_output_truncation():
    long_answer = "function build() { return 1; }\n" * 300
    with mock.patch.object(bridge_module, "_create_session", return_value="ses_x"), \
         mock.patch.object(bridge_module, "_send_message", return_value=long_answer):
        result = bridge_module._chat_payload({
            "model": brain.PRIMARY_MODEL,
            "messages": [{"role": "user", "content": "write program"}],
            "temperature": 0,
            "max_tokens": 1,
        })
    assert result["choices"][0]["message"]["content"] == long_answer
    assert bridge_module.state["last_requested_max_tokens"] == 1


def test_supervisor_start_stop_contract():
    stack_src = (ROOT / "voyager_stack.py").read_text(encoding="utf-8")
    for marker in (
        "_ensure_naldo_brain",
        "_brain_healthy",
        "_stop_brain",
        "brain_status",
        "_terminate_naldo_child",
        "ensure_opencode_installed",
        "NALDO_BRIDGE_BASE",
        "NALDO_BRAIN_PROFILE",
        "NALDO_PRIMARY_MODEL",
        "opencode_serve_command",
        "opencode_subprocess_env",
    ):
        assert marker in stack_src, f"supervisor marker missing: {marker}"
    assert "without Granite fallback" in stack_src or "without granite" in stack_src.lower()
    tree = ast.parse(stack_src)
    methods = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef) and node.name == "VoyagerStack":
            for child in node.body:
                if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    methods.add(child.name)
    for required in ("_ensure_naldo_brain", "_brain_healthy", "_stop_brain", "brain_status", "stop", "start_async"):
        assert required in methods
    assert "self.stop_event.wait(7)" in stack_src
    assert "OpenCode/bridge died mid-run; stopping Naldo immediately" in stack_src
    assert "opencode_serve_command(opencode_binary" in stack_src

    runner_src = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
    assert "Naldo OpenCode brain module unavailable" in runner_src
    assert "except Exception as exc" in runner_src


def test_supervisor_stop_terminates_all_children():
    import voyager_stack as stack_module

    stack = stack_module.VoyagerStack()
    naldo = SimpleNamespace(poll=lambda: None, terminate=lambda: setattr(naldo, "t", True), kill=lambda: None, wait=lambda timeout=None: 0)
    bridge = SimpleNamespace(poll=lambda: None, terminate=lambda: setattr(bridge, "t", True), kill=lambda: None, wait=lambda timeout=None: 0)
    opencode = SimpleNamespace(poll=lambda: None, terminate=lambda: setattr(opencode, "t", True), kill=lambda: None, wait=lambda timeout=None: 0)
    stack.proc = naldo  # type: ignore[assignment]
    stack.bridge_proc = bridge  # type: ignore[assignment]
    stack.opencode_proc = opencode  # type: ignore[assignment]
    naldo.pid = 111
    bridge.pid = 222
    opencode.pid = 333
    stack.stop_event.set()
    with mock.patch.object(stack_module, "_wait_tcp", return_value=True):
        with mock.patch.object(stack, "brain_status", return_value={"ok": True}):
            with mock.patch.object(stack, "_mineflayer_status", return_value={"ready": False}):
                status = stack.stop()
    assert getattr(naldo, "t", False) is True
    assert getattr(bridge, "t", False) is True
    assert getattr(opencode, "t", False) is True
    assert status["state"] == "stopped"


def test_gemini_bot_untouched():
    stack_src = (ROOT / "gemini_bot_stack.py").read_text(encoding="utf-8")
    assert "7863" in stack_src
    assert "gemini-3.5-flash-lite" in stack_src
    assert "opencode" not in stack_src.lower()
    assert "7864" not in stack_src
    assert "7865" not in stack_src


if __name__ == "__main__":
    test_primary_and_fallbacks_are_all_opencode_free()
    test_no_granite_gemini_fallback_anywhere()
    test_brain_lock_requires_opencode_profile_base_and_models()
    test_bridge_formats_messages_without_confusing_roles()
    test_bridge_extracts_only_text_parts()
    test_sessions_created_with_denied_permissions()
    test_loopback_only()
    test_no_credential_dependency_or_leak()
    test_supervisor_start_stop_contract()
    test_gemini_bot_untouched()
    print("PASS naldo opencode")