| from __future__ import annotations |
|
|
| import json |
| from typing import Any |
|
|
| import pytest |
|
|
| from world_simulator.config import ( |
| ConnectorConfig, |
| GameConfig, |
| NpcConfig, |
| ServerConfig, |
| SimulationConfig, |
| WorldConfig, |
| ) |
| from world_simulator.domain import Vec3 |
| from world_simulator.simulation.god_console import ( |
| GodCommandPlan, |
| GodConsole, |
| GodEdit, |
| apply_god_command_plan, |
| ) |
| from world_simulator.simulation.spawning import create_world |
|
|
|
|
| def test_god_console_parses_tool_calls_and_applies_npc_edits() -> None: |
| world = create_world(_config(npc_count=2)) |
| ada, boris = world.npcs |
| ada.health = -3 |
| ada.intention = "dead" |
| boris.position = Vec3(x=1.0, y=0.0, z=1.0) |
|
|
| captured_request: dict[str, Any] = {} |
|
|
| def fake_chat_completer(request: dict[str, Any]) -> dict[str, Any]: |
| captured_request.update(request) |
| return { |
| "choices": [ |
| { |
| "message": { |
| "content": "Ada has been rewritten by divine command.", |
| "tool_calls": [ |
| _tool_call("set_npc_health", {"npc_id": ada.id, "health": 100}), |
| _tool_call( |
| "set_npc_attack_damage", |
| {"npc_id": ada.id, "attack_damage": 50}, |
| ), |
| _tool_call( |
| "set_npc_personality", |
| {"npc_id": ada.id, "personality": "ruthless hunter"}, |
| ), |
| _tool_call( |
| "set_npc_directive", |
| { |
| "npc_id": ada.id, |
| "directive": "Hunt every visible NPC until nobody remains.", |
| }, |
| ), |
| _tool_call( |
| "set_npc_position", |
| {"npc_id": boris.id, "x": 999.0, "z": -999.0}, |
| ), |
| ], |
| } |
| } |
| ] |
| } |
|
|
| console = GodConsole( |
| ConnectorConfig( |
| type="openai_compatible", |
| base_url="http://example.test/v1", |
| model="test-model", |
| ), |
| chat_completer=fake_chat_completer, |
| ) |
|
|
| plan = console.propose(world, "Ada now wants to kill everyone and has 50 damage.") |
| applied = apply_god_command_plan(world, plan) |
|
|
| assert captured_request["model"] == "test-model" |
| assert captured_request["temperature"] == 0.6 |
| assert captured_request["top_p"] == 0.95 |
| assert "tool_choice" not in captured_request |
| assert plan.source == "openai_compatible" |
| assert plan.summary == "Ada has been rewritten by divine command." |
| assert applied == [ |
| "Ada health set to 100.", |
| "Ada damage set to 50.", |
| "Ada personality set.", |
| "Ada directive set.", |
| "Boris moved.", |
| ] |
| assert ada.health == 100 |
| assert ada.intention == "idle" |
| assert ada.attack_damage == 50 |
| assert ada.personality == "ruthless hunter" |
| assert ada.god_directive == "Hunt every visible NPC until nobody remains." |
| assert ada.memory[-1].text == "God directive: Hunt every visible NPC until nobody remains." |
| assert boris.position == Vec3(x=40.0, y=0.0, z=-40.0) |
|
|
|
|
| def test_apply_god_command_plan_can_kill_and_remember() -> None: |
| world = create_world(_config(npc_count=1)) |
| ada = world.npcs[0] |
|
|
| applied = apply_god_command_plan( |
| world, |
| GodCommandPlan( |
| source="test", |
| summary="", |
| edits=[ |
| GodEdit(name="set_npc_health", arguments={"npc_id": ada.id, "health": -1}), |
| GodEdit( |
| name="add_npc_memory", |
| arguments={"npc_id": ada.id, "memory": "A divine warning echoed."}, |
| ), |
| ], |
| ), |
| ) |
|
|
| assert applied == ["Ada health set to -1.", "Ada memory added."] |
| assert ada.intention == "dead" |
| assert ada.memory[-1].text == "A divine warning echoed." |
|
|
|
|
| def test_god_console_requires_configured_api_key_env(monkeypatch: pytest.MonkeyPatch) -> None: |
| monkeypatch.delenv("WORLD_SIMULATOR_MISSING_TEST_KEY", raising=False) |
|
|
| with pytest.raises(ValueError, match="WORLD_SIMULATOR_MISSING_TEST_KEY"): |
| GodConsole( |
| ConnectorConfig( |
| type="openai_compatible", |
| base_url="https://api.openai.com/v1", |
| model="your-tool-capable-model", |
| api_key_env="WORLD_SIMULATOR_MISSING_TEST_KEY", |
| ) |
| ) |
|
|
|
|
| def _tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: |
| return { |
| "type": "function", |
| "function": { |
| "name": name, |
| "arguments": json.dumps(arguments), |
| }, |
| } |
|
|
|
|
| def _config(*, npc_count: int) -> GameConfig: |
| return GameConfig( |
| world=WorldConfig(width=80, depth=80, terrain="plain_green", seed=42), |
| npcs=NpcConfig(count=npc_count), |
| simulation=SimulationConfig(tick_ms=500), |
| server=ServerConfig(host="127.0.0.1", port=8000), |
| ) |
|
|