"""Stability of autonomous operation: bounded status reads, restart backoff, and a single configurable local LLM path for Naldo_Bot. Offline-safe: pure stdlib plus text-level assertions on the deployment files. """ from __future__ import annotations import re 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)) from voyager_stack import restart_backoff_after, tail_lines # noqa: E402 import gemini_bot_stack as gemini_module # noqa: E402 import minecraft_stack as mc_stack # noqa: E402 from minecraft_stack import _tail_lines as mc_tail_lines # noqa: E402 def test_tail_lines_matches_full_read_on_small_logs(tmp_path): path = tmp_path / "voyager.log" expected = [f"line {i} [llm_metric] payload" for i in range(500)] path.write_text("\n".join(expected) + "\n", encoding="utf-8") assert tail_lines(path) == expected assert mc_tail_lines(path, 500) == expected def test_tail_lines_is_bounded_on_long_logs(tmp_path): path = tmp_path / "voyager.log" total = 60000 with path.open("w", encoding="utf-8") as fh: for i in range(total): fh.write(f"line {i}\n") window = tail_lines(path, 20000) assert len(window) == 20000 assert window[0] == f"line {total - 20000}" assert window[-1] == f"line {total - 1}" assert tail_lines(tmp_path / "missing.log") == [] assert mc_tail_lines(tmp_path / "missing.log") == [] def test_restart_backoff_doubles_on_quick_crash_and_resets_when_healthy(): assert restart_backoff_after(5, 12.0) == 10 assert restart_backoff_after(60, 30.0) == 120 assert restart_backoff_after(120, 1.0) == 120 assert restart_backoff_after(120, 3600.0) == 5 assert restart_backoff_after(120, 600.0) == 5 assert restart_backoff_after(10, 599.9) == 20 def test_both_bots_share_the_same_backoff_policy(): assert gemini_module.restart_backoff_after is restart_backoff_after assert gemini_module.tail_lines is tail_lines def test_naldo_uses_only_the_configurable_local_endpoint(): stack = (ROOT / "voyager_stack.py").read_text(encoding="utf-8") # Naldo generative brain is loopback OpenCode; embeddings live in the separate # CPU model Space because this Space intentionally runs server-only. assert "NALDO_BRIDGE_BASE" in stack assert "NALDO_BRAIN_PROFILE" in stack assert "NALDO_PRIMARY_MODEL" in stack assert "naldo-opencode" in stack or "NALDO_BRAIN_PROFILE" in stack assert "granite-4.0-h-tiny" not in stack runner = (ROOT / "voyager_runner.py").read_text(encoding="utf-8") assert "http://127.0.0.1:7864/v1" in runner assert "NALDO_BRAIN_PROFILE" in runner assert "from naldo_opencode import" in runner assert "naldo-granite" not in runner assert "voyager.learn(reset_env=False)" in runner def test_no_new_direct_llm_sdk_dependency(): requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8").lower() assert "google-generativeai" not in requirements assert "google-genai" not in requirements assert "anthropic" not in requirements assert "openai==0.28.1" in requirements assert "llama-cpp-python" in requirements def test_local_endpoint_stays_cpu_only(): app = (ROOT / "app.py").read_text(encoding="utf-8") assert "n_gpu_layers=0" in app assert 'os.environ["CUDA_VISIBLE_DEVICES"] = ""' in app def test_acceptance_script_defaults_match_this_deployment(): agent = (ROOT / "tests" / "test_agent.py").read_text(encoding="utf-8") app = (ROOT / "app.py").read_text(encoding="utf-8") readme = (ROOT / "README.md").read_text(encoding="utf-8") base = re.search(r'BASE = os.environ.get\("BASE", "(.*?)"\)', agent).group(1) model = re.search(r'MODEL = os.environ.get\("MODEL", "(.*?)"\)', agent).group(1) assert base in readme, "acceptance BASE default must be the documented Space URL" assert base.endswith("/v1") assert model == "granite-4.0-h-tiny" assert f'"{model}"' in app or f"'{model}'" in app or model in app def test_boot_counters_survive_tail_window_and_reset_on_new_boot(tmp_path): import inspect # Wiring: every new minecraft boot must reset the counters. assert "_reset_boot_counters" in inspect.getsource( mc_stack.MinecraftStack._start_fabric ) stack = mc_stack.MinecraftStack() empty_log = tmp_path / "stack.log" empty_log.write_text("log line without markers\n", encoding="utf-8") with mock.patch.object(mc_stack, "_log", lambda *args, **kwargs: None), \ mock.patch.object(mc_stack, "LOG_PATH", empty_log): stack._pump("minecraft", SimpleNamespace(stdout=[ "Starting minecraft server version 1.19\n", "Failed to save player data for Naldo_Bot\n", 'java.lang.NullPointerException: Cannot read field "respawnDimension" because "player" is null\n', "Failed to save player data for IffyBinkie6134\n", 'Done (1.234s)! For help, type "help"\n', ])) assert stack.failed_saves_since_boot == 2 assert stack.respawn_npe_since_boot == 1 stack._pump("viaproxy", SimpleNamespace(stdout=[ "Failed to save player data for x\n", ])) assert stack.failed_saves_since_boot == 2 assert stack.respawn_npe_since_boot == 1 # The tail window holds none of those lines, yet the counters behind # the *_since_boot contract must still report them. report = stack.status() assert report["player_save"]["failed_saves_since_boot"] == 2 assert report["player_save"]["respawn_npe_since_boot"] == 1 stack._reset_boot_counters() report = stack.status() assert report["player_save"]["failed_saves_since_boot"] == 0 assert report["player_save"]["respawn_npe_since_boot"] == 0 if __name__ == "__main__": import tempfile with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) test_tail_lines_matches_full_read_on_small_logs(tmp_path) test_tail_lines_is_bounded_on_long_logs(tmp_path) test_boot_counters_survive_tail_window_and_reset_on_new_boot(tmp_path) test_restart_backoff_doubles_on_quick_crash_and_resets_when_healthy() test_both_bots_share_the_same_backoff_policy() test_naldo_uses_only_the_configurable_local_endpoint() test_no_new_direct_llm_sdk_dependency() test_local_endpoint_stays_cpu_only() test_acceptance_script_defaults_match_this_deployment() print("PASS autonomy stability")