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