tests/test_voyager_patch_application.py

"""Run the real boot patcher against a clean upstream tree and attest it.

This is the end-to-end proof that the patch layer no longer substitutes a
decision method: the patcher is executed, then the attestation is run on the
resulting tree.
"""
from __future__ import annotations

import os
import shutil
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SIM = Path(os.environ.get("PATCH_SIM_ROOT", r"C:\Users\PCGAME~1\AppData\Local\Temp\kilo\patchsim"))
UPSTREAM_HOME = Path(
    os.environ.get("VOYAGER_UPSTREAM_REF")
    or r"C:\Users\PCGAME~1\AppData\Local\Temp\kilo\voyager-upstream-ref"
)

sys.path.insert(0, str(ROOT))

import voyager_runtime_contract as contract  # noqa: E402
import voyager_stack as stack_module  # noqa: E402


def _without_comments(source: str) -> str:
    return "\n".join(
        line for line in source.splitlines() if not line.strip().startswith("//")
    )


def _fresh_tree() -> Path:
    live = SIM / "Voyager"
    if live.exists():
        shutil.rmtree(live)
    live.mkdir(parents=True)
    shutil.copytree(UPSTREAM_HOME / "voyager", live / "voyager")
    return live


def test_patcher_runs_and_attestation_passes(monkeypatch):
    live = _fresh_tree()
    monkeypatch.setattr(stack_module, "ROOT", SIM)
    monkeypatch.setattr(stack_module, "SRC", live)
    stack = stack_module.VoyagerStack()
    stack._patch_runtime_source()

    # The snapshot must exist and be pristine.
    reference = live / contract.REFERENCE_DIR / "voyager/agents/action.py"
    expected = live / contract.REFERENCE_DIR / "voyager/agents/curriculum.py"
    assert reference.exists() and expected.exists()
    assert expected.read_text(encoding="utf-8") == (
        UPSTREAM_HOME / "voyager/agents/curriculum.py"
    ).read_text(encoding="utf-8")

    # The patched tree must still contain the official action parser.
    action = (live / "voyager/agents/action.py").read_text(encoding="utf-8")
    assert '"\\n\\n".join(function["body"] for function in functions)' in action
    assert "canonical_place_code" not in action
    curriculum = (live / "voyager/agents/curriculum.py").read_text(encoding="utf-8")
    assert "choice_id" not in curriculum
    assert "generate_affordances" not in curriculum
    assert "run_qa_step2_answer_questions" in curriculum, "official QA context removed"

    # Every action.py patch must be present in the tree that actually runs. A
    # str.replace that misses its marker leaves the file byte-identical, and the
    # Space then boots looking healthy on a pristine upstream agent.
    action_markers = {
        "base skills": '"creativeGiveItem",',
        "max_tokens": "max_tokens=2048",
        "error bound": "error_messages[-4:]",
        "chat-log bound": "chat_messages[-6:]",
        "context bound": "{context[:1400]}",
        "critique bound": "{critique[:1200]}",
    }
    for what, marker in action_markers.items():
        assert marker in action, f"action {what} patch did not apply"
    # The previous program must reach the model whole. Truncating it hid the
    # function the model had just declared, so it redeclared it and upstream's
    # Babel rejected the duplicate identifier.
    assert "Code from the last round" in action
    assert "code[:3200]" not in action, "previous code is still truncated"
    # The official Action contract is untouched by those integration bounds.
    assert "max_tokens" not in (
        UPSTREAM_HOME / "voyager/agents/action.py"
    ).read_text(encoding="utf-8"), "upstream action.py is expected to set no max_tokens"

    # No local build template may be installed. Comments may still explain the
    # removal, so compare against executable code only.
    place = _without_comments(
        (live / "voyager/control_primitives/placeItem.js").read_text(encoding="utf-8")
    )
    assert "buildProject" not in place
    assert "_voyagerProjectBlueprint" not in place
    assert not (live / "voyager/control_primitives_context/buildProject.js").exists()
    assert not (live / "voyager/control_primitives_context/placeBlocks.js").exists()

    # The official critic prompt must survive untouched.
    critic = (live / "voyager/prompts/critic.txt").read_text(encoding="utf-8")
    assert len(critic.splitlines()) > 50

    # The world is a flat resource map, so mining must be able to reach the blocks
    # under the bot's feet. The first attempt used `Math.floor(bot.y) - 2` for
    # that, but a floor measured FROM THE BOT travels down with it: it always
    # excluded the next layer, so iron at y=48 stayed visible and unmineable while
    # the bot stood at y=63. The floor must be an absolute depth budget, and the
    # primitive must be able to dig toward a layer it knows is below.
    mine = _without_comments(
        (live / "voyager/control_primitives/mineBlock.js").read_text(encoding="utf-8")
    )
    assert "MAX_DIG_DEPTH" in mine
    assert "const searchFloorY = Math.max(" in mine
    assert "Math.floor(bot.entity.position.y) - MAX_DIG_DEPTH" in mine
    assert "const playerFloorY = Math.floor(bot.entity.position.y) - 2;" not in mine
    assert "descendToward" in mine
    assert "steps < MAX_DIG_DEPTH" in mine
    assert 'under.name !== "bedrock"' in mine

    # The pathfinder must keep its upstream digging ability, otherwise a bot that
    # mines its own shaft can never path out of it again.
    index = (live / "voyager/env/mineflayer/index.js").read_text(encoding="utf-8")
    assert "movements.canDig = false" not in _without_comments(index)

    # The learning loop and recovery integration fixes must still be present.
    engine = (live / "voyager/voyager.py").read_text(encoding="utf-8")
    assert "self.skill_manager.add_new_skill(info)" in engine
    assert "deterministic_action_result" not in engine
    assert "traceback.print_exc()" in engine


def test_attestation_accepts_the_patched_tree(monkeypatch):
    live = _fresh_tree()
    monkeypatch.setattr(stack_module, "ROOT", SIM)
    monkeypatch.setattr(stack_module, "SRC", live)
    stack_module.VoyagerStack()._patch_runtime_source()

    # Import the patched agents directly. Their third-party dependencies
    # (gymnasium, langchain, chromadb) are not installed here, so the classes
    # are built from their source the same way the attestation reads them.
    from test_voyager_official_contract import _load_class

    engine_cls = _load_class(live / "voyager/voyager.py", "Voyager")
    engine = engine_cls.__new__(engine_cls)
    for relative, class_name, attribute in (
        ("voyager/agents/curriculum.py", "CurriculumAgent", "curriculum_agent"),
        ("voyager/agents/action.py", "ActionAgent", "action_agent"),
        ("voyager/agents/critic.py", "CriticAgent", "critic_agent"),
        ("voyager/agents/skill.py", "SkillManager", "skill_manager"),
    ):
        cls = _load_class(live / relative, class_name)
        setattr(engine, attribute, cls.__new__(cls))

    monkeypatch.setattr(
        contract.subprocess,
        "check_output",
        lambda *a, **k: contract.UPSTREAM_COMMIT,
    )
    attestation = contract.attest_original_runtime(engine, live)
    assert attestation["checkout_verified"] is True
    assert "action.process_ai_message" in attestation["decision_methods_verified"]
    assert "critic.check_task_success" in attestation["decision_methods_verified"]
    assert set(attestation["declared_adaptations"]) == {"curriculum.propose_next_ai_task"}