File size: 6,419 Bytes
f420896 aa8772a f420896 aa8772a f420896 aa8772a f420896 aa8772a f420896 aa8772a f420896 | 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 | """Every name the runtime touches must resolve.
Why this exists: a retry loop was added to voyager_runner.py that called
time.sleep() without importing time. The tests at the time only asserted that
certain strings appeared in the file, so they passed while the new code raised
NameError on its first use -- inside the except block, which turned a retryable
provider error into a hard crash. Gemini_Bot's restart count climbed 1 -> 4 ->
8 -> 9 and the cause was invisible.
pyflakes reports "undefined name", which is that exact failure class. Skipped
rather than failed when pyflakes is unavailable, so the suite still runs in a
bare environment.
"""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
# Modules this deployment maintains. Upstream files are excluded: they are
# pinned and must not be edited, so a warning there is not actionable.
MAINTAINED = (
"app.py",
"minecraft_stack.py",
"voyager_stack.py",
"voyager_runner.py",
"voyager_verification.py",
"voyager_learning.py",
"voyager_metrics.py",
"voyager_creative_critic.py",
"voyager_runtime_contract.py",
"gemini_bot_stack.py",
"gemini_voyager_proxy.py",
"gemini_bot_logic.js",
)
def _contract():
sys.path.insert(0, str(ROOT))
import voyager_runtime_contract
return voyager_runtime_contract
def _tmp_home() -> str:
import tempfile
return tempfile.mkdtemp(prefix="exit-report-")
def _contract():
sys.path.insert(0, str(ROOT))
import voyager_runtime_contract
return voyager_runtime_contract
def _tmp_home() -> str:
import tempfile
return tempfile.mkdtemp(prefix="exit-report-")
def _pyflakes() -> str | None:
try:
import pyflakes # noqa: F401
except ImportError:
return None
return sys.executable
def test_no_undefined_names_in_the_maintained_modules():
interpreter = _pyflakes()
if interpreter is None:
pytest.skip("pyflakes is not installed; cannot check for undefined names")
targets = [str(ROOT / name) for name in MAINTAINED if (ROOT / name).suffix == ".py"]
assert targets, "no Python modules to check"
result = subprocess.run(
[interpreter, "-m", "pyflakes", *targets],
capture_output=True,
text=True,
)
undefined = [
line
for line in result.stdout.splitlines()
if "undefined name" in line or "undefined local" in line
]
assert not undefined, (
"a name used at runtime does not resolve; this raises at execution time, "
"not at import time, so only a real call would surface it:\n "
+ "\n ".join(undefined)
)
def test_the_retry_loop_imports_what_it_uses():
# The specific regression, asserted directly as well as by the check above.
source = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
assert "time.sleep(delay)" in source
head = source[: source.index("VOYAGER_HOME = Path(")]
assert "\nimport time\n" in head, "voyager_runner uses time.sleep but does not import time"
def test_the_retry_sleep_is_inside_the_except_block():
# If the sleep sat outside the except, the loop would not back off.
source = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
start = source.index("while True:")
end = source.index("finally:", start)
body = source[start:end]
assert "except Exception as exc:" in body
assert body.index("except Exception as exc:") < body.index("time.sleep(delay)")
assert body.index("time.sleep(delay)") < body.index("delay = min(delay * 2, delay_cap)")
def test_a_crash_is_recorded_so_the_next_restart_is_diagnosable():
# The HF log stream served a frozen window and the Gemini status exposes no
# log tail, so 'runner=1' was the only symptom available.
contract = _contract()
home = Path(_tmp_home()) / "Voyager"
home.mkdir(parents=True, exist_ok=True)
try:
raise RuntimeError("provider rejected sk-abcdefghijklmnop with 401")
except RuntimeError as exc:
record = contract.record_exit(home, "Gemini_Bot", exc)
path = home.parent / "last-exit-Gemini_Bot.json"
assert path.exists(), "the report must land beside the checkout"
on_disk = json.loads(path.read_text(encoding="utf-8"))
assert on_disk["bot_username"] == "Gemini_Bot"
assert on_disk["pid"] > 0
assert on_disk["error"].startswith("RuntimeError:")
assert "401" in on_disk["error"], "the message must survive"
# It is published through a status endpoint, so it must not carry a credential.
assert "sk-abcdefghijklmnop" not in json.dumps(on_disk)
assert "<redacted>" in on_disk["error"]
assert any("RuntimeError" in line for line in on_disk["traceback"]), (
"the traceback must identify the failure"
)
# And it must be readable back through the contract.
assert contract.read_exit_report(home, "Gemini_Bot")["error"] == record["error"]
def test_each_bot_gets_its_own_report():
contract = _contract()
root = Path(_tmp_home())
(root / "Voyager").mkdir(parents=True, exist_ok=True)
contract.record_exit(root / "Voyager", "Naldo_Bot", ValueError("a"))
contract.record_exit(root / "Voyager", "Gemini_Bot", ValueError("b"))
assert contract.read_exit_report(root / "Voyager", "Naldo_Bot")["error"].endswith("a")
assert contract.read_exit_report(root / "Voyager", "Gemini_Bot")["error"].endswith("b")
def test_a_write_failure_does_not_mask_the_original_error(tmp_path):
contract = _contract()
# Point the report at a path that cannot be created.
blocker = tmp_path / "blocked"
blocker.write_text("not a directory", encoding="utf-8")
assert contract.record_exit(blocker / "Voyager", "Naldo_Bot", ValueError("x")) == {}
def test_the_exit_report_is_written_outside_the_pinned_checkout():
# A runtime file inside the checkout would look like an undeclared addition
# to the divergence check and fail the next boot.
import sys as _sys
_sys.path.insert(0, str(ROOT))
import voyager_runtime_contract as contract
home = ROOT / "work" / "patched3" / "Voyager"
path = contract.exit_report_path(home, "Gemini_Bot")
assert path.parent == home.parent, "the report must sit beside the checkout, not inside it"
assert path.name == "last-exit-Gemini_Bot.json"
|