File size: 8,645 Bytes
717e39b 008c4fa 717e39b 008c4fa 717e39b 008c4fa 717e39b 008c4fa 717e39b 008c4fa 717e39b 008c4fa 717e39b 008c4fa 717e39b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | """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", <default>))`, 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"
)
|