Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| from world_simulator.domain import ( | |
| Beast, | |
| Npc, | |
| NpcConversationContext, | |
| ResourceNode, | |
| Terrain, | |
| Vec3, | |
| WorldEvent, | |
| WorldState, | |
| ) | |
| from world_simulator.simulation.connectors.base import NpcDirective | |
| from world_simulator.simulation.mechanics import BLOCK_SIZE | |
| from world_simulator.simulation.survival import ( | |
| add_episode, | |
| apply_action_effects, | |
| apply_survival_tick, | |
| build_memory_summary, | |
| build_perception, | |
| resolve_call_for_help, | |
| select_goal, | |
| ) | |
| def _npc( | |
| npc_id: str = "npc-001", | |
| name: str = "Ada", | |
| *, | |
| x: float = 0.0, | |
| z: float = 0.0, | |
| **state: object, | |
| ) -> Npc: | |
| return Npc(id=npc_id, name=name, role="forager", position=Vec3(x=x, y=0.0, z=z), **state) # type: ignore[arg-type] | |
| def _world( | |
| npcs: list[Npc], | |
| *, | |
| beasts: list[Beast] | None = None, | |
| resource_nodes: list[ResourceNode] | None = None, | |
| active_events: list[WorldEvent] | None = None, | |
| tick: int = 0, | |
| ) -> WorldState: | |
| return WorldState( | |
| tick=tick, | |
| seed=42, | |
| terrain=Terrain(kind="plain_green", width=80, depth=80), | |
| npcs=npcs, | |
| beasts=beasts or [], | |
| resource_nodes=resource_nodes or [], | |
| active_events=active_events or [], | |
| ) | |
| def test_add_episode_saves_structured_fields() -> None: | |
| npc = _npc() | |
| add_episode( | |
| npc, | |
| 7, | |
| "attack", | |
| "beast_1", | |
| "beast_1 attacked me", | |
| target_id=npc.id, | |
| subject_kind="beast", | |
| perspective="recipient", | |
| tags=["danger", "beast"], | |
| weight=0.9, | |
| ) | |
| episode = npc.structured_memory.episodes[0] | |
| assert episode.tick == 7 | |
| assert episode.kind == "attack" | |
| assert episode.actor_id == "beast_1" | |
| assert episode.target_id == npc.id | |
| assert episode.subject_kind == "beast" | |
| assert episode.perspective == "recipient" | |
| assert episode.summary == "beast_1 attacked me" | |
| assert episode.tags == ["danger", "beast"] | |
| assert episode.emotional_weight == 0.9 | |
| assert npc.relationships["beast_1"] == -1.0 | |
| def test_add_episode_trims_to_twenty_by_emotional_weight() -> None: | |
| npc = _npc() | |
| for index in range(25): | |
| add_episode( | |
| npc, | |
| index, | |
| "observe", | |
| f"actor-{index}", | |
| f"event {index}", | |
| weight=index / 24, | |
| ) | |
| assert len(npc.structured_memory.episodes) == 20 | |
| assert min(episode.emotional_weight for episode in npc.structured_memory.episodes) >= 5 / 24 | |
| assert all(episode.summary != "event 0" for episode in npc.structured_memory.episodes) | |
| def test_build_memory_summary_includes_last_n_episodes() -> None: | |
| npc = _npc() | |
| add_episode(npc, 1, "observe", "beast_1", "first memory", tags=["beast"], weight=0.4) | |
| add_episode( | |
| npc, | |
| 2, | |
| "transfer", | |
| "npc-002", | |
| "second memory", | |
| object_id="food", | |
| subject_kind="npc", | |
| perspective="recipient", | |
| tags=["food", "help", "trade"], | |
| weight=0.5, | |
| ) | |
| add_episode( | |
| npc, | |
| 3, | |
| "attack", | |
| "beast_1", | |
| "third memory", | |
| subject_kind="beast", | |
| perspective="recipient", | |
| tags=["danger", "beast"], | |
| weight=0.9, | |
| ) | |
| summary = build_memory_summary(npc, last_n=2, current_tick=4) | |
| assert "first memory" not in summary | |
| assert "second memory" in summary | |
| assert "third memory" in summary | |
| assert "1 ticks ago" in summary | |
| def test_transfer_food_increases_trust() -> None: | |
| giver = _npc("npc-001", "Ada", inventory_food=3) | |
| taker = _npc("npc-002", "Boris", x=2.0, inventory_food=0) | |
| world = _world([giver, taker]) | |
| summary = apply_action_effects(giver, "transfer", world) | |
| assert summary == "sharing food with Boris" | |
| assert giver.relationships[taker.id] == 0.2 | |
| assert taker.relationships[giver.id] > 0 | |
| assert taker.structured_memory.episodes[-1].kind == "transfer" | |
| assert "food" in taker.structured_memory.episodes[-1].tags | |
| def test_relationship_updates_on_steal() -> None: | |
| thief = _npc("npc-001", "Ada") | |
| victim = _npc("npc-002", "Boris", x=2.0, inventory_food=1) | |
| world = _world([thief, victim]) | |
| summary = apply_action_effects( | |
| thief, | |
| "transfer", | |
| world, | |
| directive=NpcDirective(npc_id=thief.id, action="transfer", take=True), | |
| ) | |
| assert summary == "stealing from Boris" | |
| assert victim.relationships[thief.id] == -0.5 | |
| assert victim.structured_memory.episodes[-1].kind == "steal" | |
| assert victim.structured_memory.episodes[-1].perspective == "recipient" | |
| def test_beast_attack_sets_enemy_relationship() -> None: | |
| victim = _npc(x=1.5, z=0.0) | |
| beast = Beast("beast_1", Vec3(0.0, 0.0, 0.0)) | |
| world = _world([victim], beasts=[beast]) | |
| apply_survival_tick(world) | |
| assert victim.relationships["beast_1"] == -1.0 | |
| episode = victim.structured_memory.episodes[-1] | |
| assert episode.kind == "attack" | |
| assert episode.perspective == "recipient" | |
| assert "beast" in episode.tags | |
| def test_witness_attack_writes_attack_episode() -> None: | |
| victim = _npc("npc-001", "Ada", x=1.5, z=0.0) | |
| witness = _npc("npc-002", "Boris", x=5.0, z=0.0) | |
| beast = Beast("beast_1", Vec3(0.0, 0.0, 0.0)) | |
| world = _world([victim, witness], beasts=[beast]) | |
| apply_survival_tick(world) | |
| episode = witness.structured_memory.episodes[-1] | |
| assert episode.kind == "attack" | |
| assert episode.actor_id == "beast_1" | |
| assert episode.target_id == victim.id | |
| assert episode.perspective == "witness" | |
| assert {"danger", "beast"}.issubset(episode.tags) | |
| def test_call_for_help_writes_communicate_episode_without_trust_gain() -> None: | |
| caller = _npc("npc-001", "Ada", fear=70.0) | |
| listener = _npc("npc-002", "Boris", x=4.0, inventory_weapon=0) | |
| world = _world([caller, listener]) | |
| summary = apply_action_effects( | |
| caller, | |
| "speak", | |
| world, | |
| directive=NpcDirective( | |
| npc_id=caller.id, | |
| action="speak", | |
| communication_intent="help_request", | |
| ), | |
| ) | |
| assert summary == "calling for help" | |
| caller_episode = caller.structured_memory.episodes[-1] | |
| listener_episode = listener.structured_memory.episodes[-1] | |
| assert caller_episode.kind == "communicate" | |
| assert listener_episode.kind == "communicate" | |
| assert {"help", "danger"}.issubset(listener_episode.tags) | |
| assert caller.relationships.get(listener.id, 0.0) == 0.0 | |
| assert listener.relationships.get(caller.id, 0.0) == 0.0 | |
| def test_call_for_help_recruits_armed_trusted_helper() -> None: | |
| caller = _npc("npc-001", "Ada") | |
| helper = _npc("npc-002", "Boris", x=4.0, inventory_weapon=1) | |
| helper.relationships[caller.id] = 0.5 | |
| world = _world([caller, helper]) | |
| helpers = resolve_call_for_help(caller, world) | |
| assert helpers == [helper] | |
| assert helper.survival_goal == "help_ally" | |
| assert helper.help_target_id == caller.id | |
| assert helper.structured_memory.episodes[-1].kind == "communicate" | |
| assert {"help", "danger"}.issubset(helper.structured_memory.episodes[-1].tags) | |
| def test_call_for_help_ignores_unarmed_or_low_trust_npcs() -> None: | |
| caller = _npc("npc-001", "Ada") | |
| unarmed = _npc("npc-002", "Boris", x=4.0, inventory_weapon=0) | |
| low_trust = _npc("npc-003", "Cora", x=8.0, inventory_weapon=1) | |
| unarmed.relationships[caller.id] = 0.8 | |
| low_trust.relationships[caller.id] = 0.1 | |
| world = _world([caller, unarmed, low_trust]) | |
| helpers = resolve_call_for_help(caller, world) | |
| assert helpers == [] | |
| assert unarmed.help_target_id is None | |
| assert low_trust.help_target_id is None | |
| def test_nearby_beast_does_not_force_a_threat_goal() -> None: | |
| # A nearby beast no longer overrides the goal: the LLM sees the threat in the | |
| # briefing and decides for itself whether to fight, flee, or call for help. | |
| npc = _npc("npc-001", "Ada", x=8.0, inventory_weapon=1) | |
| neighbor = _npc("npc-002", "Boris", x=12.0) | |
| beast = Beast("beast_1", Vec3(0.0, 0.0, 0.0)) | |
| goal = select_goal(npc, _world([npc, neighbor], beasts=[beast])) | |
| assert not goal.startswith("survive_threat") | |
| assert goal != "engage_threat" | |
| def test_build_perception_includes_beast_in_radius() -> None: | |
| npc = _npc() | |
| beast = Beast("beast_1", Vec3(2 * BLOCK_SIZE, 0.0, 0.0)) | |
| world = _world([npc], beasts=[beast]) | |
| perception = build_perception(npc, world) | |
| assert "beast_1" in perception | |
| assert "[danger]" in perception | |
| def test_build_perception_includes_resource_nodes_in_radius() -> None: | |
| npc = _npc() | |
| node = ResourceNode("food_bush_1", "food", Vec3(2 * BLOCK_SIZE, 0.0, 0.0), amount=4) | |
| world = _world([npc], resource_nodes=[node]) | |
| perception = build_perception(npc, world) | |
| assert "food_bush_1 is 2 tiles WEST" in perception | |
| assert "[food=4]" in perception | |
| def test_build_perception_includes_nearby_relationship_status() -> None: | |
| ada = _npc("npc-001", "Ada") | |
| boris = _npc("npc-002", "Boris", x=4.0) | |
| elena = _npc("npc-003", "Elena", z=4.0) | |
| ada.relationships[boris.id] = 0.8 | |
| ada.relationships[elena.id] = -0.5 | |
| world = _world([ada, boris, elena]) | |
| perception = build_perception(ada, world) | |
| assert "Boris nearby [ally trust=0.8" in perception | |
| assert "Elena nearby [enemy trust=-0.5" in perception | |
| def test_conversation_context_keeps_last_five_turns() -> None: | |
| context = NpcConversationContext(max_turns=5) | |
| for index in range(7): | |
| context.add_turn(f"u{index}", f"a{index}") | |
| assert len(context.messages) == 10 | |
| assert context.messages[0] == {"role": "user", "content": "u2"} | |
| assert context.messages[-1] == {"role": "assistant", "content": "a6"} | |
| def test_world_event_is_injected_into_perception_when_in_radius() -> None: | |
| npc = _npc() | |
| event = WorldEvent( | |
| tick_created=0, | |
| kind="observe", | |
| description="A stranger appears near the well.", | |
| position=Vec3(4.0, 0.0, 0.0), | |
| radius=8.0, | |
| ) | |
| world = _world([npc], active_events=[event]) | |
| perception = build_perception(npc, world) | |
| assert "[EVENT] A stranger appears near the well." in perception | |