tests/test_altoclef_stack.py

"""Lifecycle and world-scope checks, without launching a Minecraft client."""
import io
import sys
import time
from pathlib import Path
from types import SimpleNamespace

import pytest

sys.path.insert(0, str(Path(__file__).parents[1]))
import altoclef_stack as module

@pytest.fixture
def runner(monkeypatch, tmp_path):
    monkeypatch.setattr(module, "ROOT", tmp_path)
    monkeypatch.delenv("ALTOCLEF_OPENCODE_MODEL", raising=False)
    monkeypatch.delenv("MC_USERNAME", raising=False)
    return module.AltoClefStack()

def response(data):
    return SimpleNamespace(raise_for_status=lambda: None, json=lambda: data)

def test_stale_connected_telemetry_is_an_error(runner, monkeypatch):
    monkeypatch.setattr(module.requests, "get", lambda *a, **k:
                        response({"connected": True, "observed_at": time.time() - 60,
                                  "heartbeat_at": time.time()}))
    with pytest.raises(RuntimeError, match="stopped updating"):
        runner._observe()

def test_loading_heartbeat_accepted_inside_grace(runner, monkeypatch):
    now = time.time()
    runner.last_started_at = now
    monkeypatch.setattr(module.requests, "get", lambda *a, **k:
                        response({"connected": False, "loading": True, "heartbeat_at": now}))
    assert runner._observe()["loading"] is True

def test_loading_without_heartbeat_is_an_error(runner, monkeypatch):
    runner.last_started_at = time.time()
    monkeypatch.setattr(module.requests, "get", lambda *a, **k:
                        response({"connected": False, "loading": True, "heartbeat_at": time.time() - 120}))
    with pytest.raises(RuntimeError, match="heartbeat stopped"):
        runner._observe()

def test_exact_initial_payload_accepted_only_in_first_window(runner, monkeypatch):
    # Java's initial /state before the first 2s scheduler tick.
    runner.last_started_at = time.time()
    monkeypatch.setattr(module.requests, "get", lambda *a, **k:
                        response({"connected": False, "loading": True, "heartbeat_at": 0}))
    assert runner._observe()["loading"] is True
    runner._bridge_seen_at = time.time() - module.FIRST_HEARTBEAT_WINDOW - 1
    with pytest.raises(RuntimeError, match="never published a heartbeat"):
        runner._observe()

def test_reconnect_allowed_bounded_after_play(runner, monkeypatch):
    now = time.time()
    runner.last_started_at = now - module.LOADING_GRACE - 10
    runner._last_connected_at = now - 60
    monkeypatch.setattr(module.requests, "get", lambda *a, **k:
                        response({"connected": False, "loading": True, "heartbeat_at": now}))
    assert runner._observe()["loading"] is True
    runner._last_connected_at = now - module.RECONNECT_GRACE - 1
    with pytest.raises(RuntimeError, match="did not reconnect"):
        runner._observe()

def test_second_boot_resets_launch_clocks(runner, monkeypatch):
    # Regression: after a crash, the previous launch leaves an old connected
    # timestamp, a stale first-heartbeat clock and an expired grace anchor.
    # The new boot must accept its fresh initial loading payload.
    now = time.time()
    runner._last_connected_at = now - module.RECONNECT_GRACE - 60
    runner._bridge_seen_at = now - module.FIRST_HEARTBEAT_WINDOW - 60
    runner.last_started_at = now - module.LOADING_GRACE - 60
    runner._last_gamer_at = now
    runner._last_mode_set = now
    runner._reset_launch_clocks()
    assert runner._bridge_seen_at is None
    assert runner._last_connected_at is None
    assert runner.last_started_at is None
    assert runner._last_gamer_at == 0.0 and runner._last_mode_set == 0.0
    runner.last_started_at = time.time()
    monkeypatch.setattr(module.requests, "get", lambda *a, **k:
                        response({"connected": False, "loading": True, "heartbeat_at": 0}))
    assert runner._observe()["loading"] is True

def test_boot_resets_launch_clocks_before_startup():
    import inspect
    source = inspect.getsource(module.AltoClefStack._boot)
    assert "_reset_launch_clocks()" in source
    assert source.index("_reset_launch_clocks()") < source.index("minecraft_ready.wait")

def _bridge_source() -> str:
    return (Path(module.__file__).parents[0] / "altoclef" / "SpaceBridge.java").read_text(encoding="utf-8")

def test_bridge_gates_manual_connect_on_readiness():
    # Pinned Yarn 1.19.4+build.2: TitleScreen class_442, DisconnectedScreen
    # class_419, ConnectScreen class_412, overlay via getOverlay. The manual
    # connect must take the current screen and refuse while a reload overlay
    # is up or vanilla is already connecting.
    source = _bridge_source()
    assert "class_442" in source and "class_419" in source
    # Pinned Yarn 1.19.4+build.2 row: METHOD emh ()Lesx; aJ method_18506
    # getOverlay. The live runtime is intermediary, so the mapped ID (not the
    # Yarn name) must be used.
    assert "method_18506" in source
    assert '"getOverlay"' not in source
    assert "connect(client, currentScreen)" in source
    assert "mc(\"class_412\").isInstance(currentScreen)" in source
    import re
    assert not re.search(r"(?<!, )connect\(client\)", source), \
        "ungated single-arg connect call remains"

def test_bridge_reports_connect_diagnostics():
    source = _bridge_source()
    assert '"connect_state"' in source and '"connect_attempts"' in source

def test_built_bridge_jar_matches_source_version():
    import json as _json
    import zipfile as _zip
    jar = Path(module.__file__).parents[0] / "altoclef" / "space-altoclef-bridge.jar"
    meta = _json.loads(_zip.ZipFile(jar).read("fabric.mod.json").decode())
    assert meta["version"] == "1.0.4"
    assert "space/minecraft/SpaceBridge.class" in _zip.ZipFile(jar).namelist()

def test_client_starts_at_title_without_autojoin_flags(runner, monkeypatch, tmp_path):
    import sys as _sys
    seen: dict = {}
    fake_launcher = SimpleNamespace(
        fabric=SimpleNamespace(install_fabric=lambda *a, **k: None),
        command=SimpleNamespace(
            get_minecraft_command=lambda *a, **k: (seen.update(a[2]), ["java"])[1]),
    )
    monkeypatch.setitem(_sys.modules, "minecraft_launcher_lib", fake_launcher)
    monkeypatch.setattr(module, "CLIENT", tmp_path / "client")
    monkeypatch.setattr(module, "MOD_CACHE", tmp_path / "cache")
    monkeypatch.setattr(module, "_download",
                        lambda url, path, sha=None: (path.parent.mkdir(parents=True, exist_ok=True),
                                                    path.write_bytes(b"x"))[1])
    assert runner._ensure_client(Path("java")) == ["java"]
    # No vanilla --server/--port: autojoin bypasses the TitleScreen, and
    # upstream AltoClef initializes ONLY there (EntryMixin on TitleScreen.init
    # -> TitleScreenEntryEvent -> onInitializeLoad). The bridge's gated
    # fallback connects after resources are done instead.
    assert "server" not in seen and "port" not in seen
    options = (tmp_path / "client" / "options.txt").read_text(encoding="utf-8")
    assert "onboardAccessibility:false" in options
    # Second ensure must not duplicate the key.
    runner._ensure_client(Path("java"))
    options = (tmp_path / "client" / "options.txt").read_text(encoding="utf-8")
    assert options.count("onboardAccessibility:false") == 1

def test_bridge_reports_uninitialized_taskchain_without_npe():
    # A null task chain during loading must surface as initialized=false with
    # diagnostics, never an NPE, and never a fabricated task_active.
    source = _bridge_source()
    assert "chain == null" in source
    assert "taskchain not yet initialized" in source

def test_survival_verified_requires_explicit_survival():
    good = {"connected": True, "error": "", "creative": False}
    assert module.AltoClefStack._survival_verified(good) is True
    # Loading payload without a creative flag is NOT Survival.
    assert module.AltoClefStack._survival_verified({"connected": False, "loading": True}) is False
    assert module.AltoClefStack._survival_verified({"connected": True, "error": "", "creative": True}) is False
    assert module.AltoClefStack._survival_verified({"connected": True, "error": "boom", "creative": False}) is False

def test_loading_past_grace_is_an_error(runner, monkeypatch):
    now = time.time()
    runner.last_started_at = now - module.LOADING_GRACE - 10
    monkeypatch.setattr(module.requests, "get", lambda *a, **k:
                        response({"connected": False, "loading": True, "heartbeat_at": now,
                                  "dispatch_error": "client thread stalled"}))
    with pytest.raises(RuntimeError, match="still not connected"):
        runner._observe()

def test_credits_evidence_survives_restart(runner, monkeypatch):
    state = {"connected": True, "entered_end": True, "credits_observed": True,
             "observed_at": time.time(), "heartbeat_at": time.time()}
    monkeypatch.setattr(module.requests, "get", lambda *a, **k: response(state))
    runner._observe()
    restored = module.AltoClefStack()
    assert restored.status()["state"] == "completed"
    assert restored.status()["goal_completed"]
    assert not restored.status()["running"]

def test_completed_goal_is_not_restarted_by_supervisor(runner, monkeypatch):
    runner.observation = {"entered_end": True, "credits_observed": True}
    runner.state = "stopped"
    monkeypatch.setattr(module.threading, "Thread", lambda *a, **k:
                        pytest.fail("completed goal launched another client"))
    assert runner.start_async()["state"] == "completed"

def test_inactive_task_is_not_completion_after_restart(runner):
    runner.observation = {"entered_end": True, "task_active": False, "dragon_seen": False}
    runner._checkpoint()
    restored = module.AltoClefStack()
    assert restored.status()["state"] == "stopped"
    assert not restored.status()["goal_completed"]

def test_boot_failures_respect_backoff(runner, monkeypatch):
    runner.state = "error"
    runner.next_retry_at = time.time() + 120
    monkeypatch.setattr(module.threading, "Thread", lambda *a, **k:
                        pytest.fail("retry started before its deadline"))
    assert runner.start_async()["state"] == "error"
    assert runner.restart_count == 0

def test_single_username_is_naldo_bot():
    assert module.USERNAME == "Naldo_Bot"

def test_default_coach_model_is_muse_spark_only(runner):
    import opencode_config
    assert runner.opencode_model == "opencode/muse-spark-1.3-contributor-free"
    assert module.COACH_MODEL == runner.opencode_model == opencode_config.MUSE_MODEL

def test_mode_assignment_only_sets_survival_for_this_player(runner, monkeypatch):
    console = io.StringIO()
    server = SimpleNamespace(stdin=console, poll=lambda: None)
    monkeypatch.setattr(module.STACK, "processes", {"minecraft": server})
    runner._ensure_survival_mode()
    written = console.getvalue()
    assert written == f"gamemode survival {module.USERNAME}\n"
    assert "give " not in written and "clear" not in written

def test_existing_inventory_is_never_touched():
    # Assisted run: no give/clear/supply helper may exist on the supervisor.
    assert not hasattr(module.AltoClefStack, "_supply_creative")
    source = Path(module.__file__).read_text(encoding="utf-8")
    assert "minecraft:{name}" not in source
    assert "CREATIVE_LOADOUT" not in source

def test_mode_is_verified_survival_in_status(runner):
    assert "Survival" in runner.status()["world_mode"]
    assert "Normal" in runner.status()["world_mode"]

def test_consult_provides_stdin_eof(runner, monkeypatch):
    seen = {}
    def fake_popen(*a, **k):
        seen.update(k)
        return SimpleNamespace(returncode=1,
                               communicate=lambda input=None, timeout=None: ("", "provider unavailable"))
    monkeypatch.setattr(module.subprocess, "Popen", fake_popen)
    monkeypatch.setattr(module, "stop_runner", lambda p, **k: None)
    client = object()
    runner.client = client
    runner._consult(Path("opencode"), {"connected": True, "task_active": True})
    assert runner.client is client
    assert runner.model_errors == 1
    assert runner.model_checks == 0
    import subprocess as stdlib_subprocess
    assert seen.get("stdin") == stdlib_subprocess.PIPE

def test_provider_failure_does_not_stop_gameplay(runner, monkeypatch):
    process = SimpleNamespace(returncode=1, communicate=lambda **k: ("", "provider unavailable"))
    monkeypatch.setattr(module.subprocess, "Popen", lambda *a, **k: process)
    monkeypatch.setattr(module, "stop_runner", lambda p, **k: None)
    client = object()
    runner.client = client
    runner._consult(Path("opencode"), {"connected": True, "task_active": True})
    assert runner.client is client
    assert runner.model_errors == 1
    assert runner.model_checks == 0