"""Lock the official Voyager constructor contract against the pinned upstream. Upstream `Voyager.__init__` ships specific defaults. This deployment passes several of them explicitly. Every value it passes must either equal the upstream default, or appear in `DOCUMENTED_DEPARTURES` with a reason. This is a semantic check: the upstream signature and the local call are both parsed with `ast`, and the values are compared. It does not pass by finding a substring in a file -- the previous version of this test did that and silently covered only three of the ~30 parameters. """ from __future__ import annotations import ast import os from pathlib import Path ROOT = Path(__file__).resolve().parents[1] _UNSET = object() def upstream_home() -> Path: """Locate the pristine checkout of the pinned upstream commit.""" candidates = [ os.environ.get("VOYAGER_UPSTREAM_REF"), r"C:\Users\PCGAME~1\AppData\Local\Temp\kilo\voyager-upstream-ref", ] for candidate in candidates: if candidate and (Path(candidate) / "voyager" / "voyager.py").exists(): return Path(candidate) raise RuntimeError( "Pristine upstream reference not found. Set VOYAGER_UPSTREAM_REF to a " "checkout of MineDojo/Voyager at " "55e45a880755d0c8c66ca7fb5fe7962ac8974f89." ) # Parameters this deployment is allowed to pass a value other than the upstream # default for. Each entry is a deliberate, reviewed decision. Anything passed # that is not listed here must equal the upstream default exactly. DOCUMENTED_DEPARTURES = { "mc_port": "FAQ: mc_port is the documented alternative to azure_login", "server_port": "FAQ: each Mineflayer process listens on one port; two bots need two ports", "openai_api_key": "credential for the substituted backend, not a behaviour parameter", "action_agent_model_name": "declared backend substitution: local Granite instead of GPT-4", "curriculum_agent_model_name": "declared backend substitution", "curriculum_agent_qa_model_name": "declared backend substitution", "critic_agent_model_name": "declared backend substitution", "skill_manager_model_name": "declared backend substitution", "ckpt_dir": "deployment path under /persist; README documents ckpt_dir as a parameter", "resume": "README: resume=True is the documented way to continue from a checkpoint", "skill_manager_retrieval_top_k": ( "FAQ: 'if you insist on using GPT-3.5, it is essential to configure it with " "skill_manager_retrieval_top_k <= 2 to reduce the context length of the prompt'" ), } # The runner's own overrides, asserted explicitly so a silent change fails here. RUNNER_ENV_FALLBACKS = { "env_wait_ticks": 20, "env_request_timeout": 600, "max_iterations": 160, "openai_api_request_timeout": 240, } def _literal(node: ast.AST): try: return ast.literal_eval(node) except Exception: return _UNSET def _env_fallback(node: ast.AST): """Return the effective default of `int(env("NAME", ))`, else _UNSET. The env default is a string and the outer call coerces it, so the coercion is applied here before the value is compared with the upstream default. """ coerce = None inner = node if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in ("int", "float", "str") and len(node.args) == 1 ): coerce = node.func.id inner = node.args[0] if ( isinstance(inner, ast.Call) and isinstance(inner.func, ast.Name) and inner.func.id == "env" and len(inner.args) >= 2 ): try: raw = ast.literal_eval(inner.args[1]) except Exception: return _UNSET if coerce in ("int", "float") and isinstance(raw, str): try: return int(raw) if coerce == "int" else float(raw) except ValueError: return _UNSET return raw return _UNSET def upstream_defaults() -> dict: source = (upstream_home() / "voyager" / "voyager.py").read_text(encoding="utf-8") for node in ast.walk(ast.parse(source)): if isinstance(node, ast.ClassDef) and node.name == "Voyager": for child in node.body: if isinstance(child, ast.FunctionDef) and child.name == "__init__": names = [a.arg for a in child.args.args][1:] # drop self defaults = child.args.defaults paired = zip(names[len(names) - len(defaults):], defaults) return {name: _literal(default) for name, default in paired} raise AssertionError("Voyager.__init__ not found in the pinned upstream file") def runner_call() -> dict: source = (ROOT / "voyager_runner.py").read_text(encoding="utf-8") for node in ast.walk(ast.parse(source)): if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "Voyager" ): return {kw.arg: kw.value for kw in node.keywords if kw.arg} raise AssertionError("Voyager(...) call not found in voyager_runner.py") def test_upstream_defaults_are_parsed_from_the_signature(): defaults = upstream_defaults() # Spot-check the documented values the README/FAQ rely on. assert defaults["action_agent_task_max_retries"] == 4 assert defaults["action_agent_show_chat_log"] is True assert defaults["skill_manager_retrieval_top_k"] == 5 assert defaults["env_wait_ticks"] == 20 assert defaults["critic_agent_mode"] == "auto" assert defaults["curriculum_agent_mode"] == "auto" assert defaults["action_agent_model_name"] == "gpt-4" assert len(defaults) > 25, "signature parse looks truncated" def test_every_passed_parameter_matches_upstream_or_is_a_documented_departure(): defaults = upstream_defaults() passed = runner_call() mismatches = [] for name, node in passed.items(): if name in DOCUMENTED_DEPARTURES: continue assert name in defaults, f"voyager_runner.py passes unknown parameter {name!r}" literal = _literal(node) if literal is not _UNSET: if literal != defaults[name]: mismatches.append(f"{name}={literal!r} (upstream default {defaults[name]!r})") continue fallback = _env_fallback(node) if fallback is _UNSET: mismatches.append(f"{name} is not a literal and is not a documented departure") elif fallback != defaults[name]: mismatches.append( f"{name} env fallback {fallback!r} (upstream default {defaults[name]!r})" ) assert not mismatches, "constructor drifted from upstream: " + "; ".join(mismatches) def test_runner_env_fallbacks_equal_the_official_defaults(): defaults = upstream_defaults() passed = runner_call() for name, expected in RUNNER_ENV_FALLBACKS.items(): assert name in passed, f"{name} is no longer passed by voyager_runner.py" fallback = _env_fallback(passed[name]) assert fallback == expected, f"{name} fallback changed to {fallback!r}" assert fallback == defaults[name], ( f"{name} fallback {fallback!r} differs from the upstream default {defaults[name]!r}" ) def test_departures_are_exactly_the_reviewed_set(): # A new departure must be added here deliberately, with a reason. assert set(DOCUMENTED_DEPARTURES) == { "mc_port", "server_port", "openai_api_key", "action_agent_model_name", "curriculum_agent_model_name", "curriculum_agent_qa_model_name", "critic_agent_model_name", "skill_manager_model_name", "ckpt_dir", "resume", "skill_manager_retrieval_top_k", } for name, reason in DOCUMENTED_DEPARTURES.items(): assert reason and len(reason) > 20, f"{name} needs a real reason" def test_parameters_left_to_the_upstream_default_are_not_passed(): # Passing a parameter with the upstream default is harmless but it hides the # intent; more importantly, omitting one is the documented way to accept the # official default. Assert the runner does not pass the behaviour switches # that upstream keeps off, so they cannot be flipped silently. passed = runner_call() for name in ("reset_placed_if_failed", "curriculum_agent_warm_up", "skill_library_dir"): assert name not in passed, ( f"{name} is passed explicitly; leave it out so the upstream default applies" )