| """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 |
| import gemini_bot_stack as gemini_module |
| import minecraft_stack as mc_stack |
| from minecraft_stack import _tail_lines as mc_tail_lines |
|
|
|
|
| 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") |
| |
| |
| 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 |
|
|
| |
| 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 |
|
|
| |
| |
| 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") |
|
|
