"""Deterministic village core loop: hunger, resources, beasts, memory, trust. The LLM may select one action/tool, but this module owns every world mutation: position, health, inventory, resource amounts, beast state, memory episodes, and relationship changes all flow through deterministic validation/effects here. Distances in this module are in world units. The task design is expressed in "tiles"; one tile is one terrain block (``BLOCK_SIZE`` world units), so tile-based thresholds are scaled by ``TILE`` below. """ from __future__ import annotations import math import os import random from world_simulator.domain import ( Beast, CannonState, CountryState, ElectionState, House, MemoryEntry, MemoryEpisode, Npc, ResourceNode, TreasuryState, Vec3, WorldEvent, WorldLogEvent, WorldState, ) from world_simulator.simulation.connectors.base import NpcDirective, TickPlan from world_simulator.simulation.directives import clear_active_directive from world_simulator.simulation.mechanics import ( ATTACK_RADIUS, BLOCK_SIZE, TALK_RADIUS, is_alive, move_away, move_toward, vec_distance, ) from world_simulator.simulation.memory import remember, strip_tick_entries from world_simulator.simulation.roles import ( canonical_action, normalize_role, ) TILE = BLOCK_SIZE # Hunger / health economy. HUNGER_RATE = 0.8 STARVATION_THRESHOLD = 70.0 SEVERE_STARVATION_THRESHOLD = 90.0 STARVATION_DAMAGE = 0.5 SEVERE_STARVATION_DAMAGE = 1.5 HUNGER_CAP = 150.0 HEAL_GOAL_HEALTH = 40 # health < this (with herbs) -> heal_self FIND_HERBS_HEALTH = 50 # health < this (without herbs) -> find_herbs EAT_GOAL_HUNGER = 60.0 REPRODUCTION_SNACK_HUNGER = 45.0 # A would-be parent only needs to be reasonably fed (not literally stuffed) to # have a child. Set well below REPRODUCTION_SNACK_HUNGER so a single meal closes # the gap, otherwise families almost never form within a demo run. REPRODUCTION_MAX_HUNGER = 25.0 # Awareness radii (world units). BEAST_ALERT_RADIUS = 6 * TILE BEAST_THREAT_RADIUS = 8 * TILE VILLAGE_THREAT_RADIUS = 10 * TILE HOUSE_THREAT_RADIUS = 12 * TILE ALLY_RADIUS = 5 * TILE WITNESS_RADIUS = 10.0 FEAR_DECAY_RADIUS = 12 * TILE GATHER_RADIUS = 1.5 * TILE TRADE_RADIUS = 2 * TILE PERCEPTION_BEAST_RADIUS = 8 * TILE PERCEPTION_RESOURCE_RADIUS = 6 * TILE PERCEPTION_NPC_RADIUS = 6 * TILE HELP_RADIUS = 6 * TILE SPEECH_RADIUS = 6 * TILE # how far ordinary speech/shouts are heard by others HOUSE_RADIUS = 2.5 * TILE HOUSE_BUILD_RADIUS = 2 * TILE HOUSE_MIN_DISTANCE = 4 * TILE # Repairing a standing-but-damaged home: cheap, fast, and the builder's top job # when beasts are chewing on the base. Distinct from rebuilding a destroyed ruin. HOUSE_REPAIR_PER_ACTION = 12.0 HOUSE_REPAIR_WOOD_COST = 1 # Fear dynamics. FEAR_ON_ATTACK = 60.0 FEAR_ON_WITNESS = 20.0 FEAR_ON_DEATH_BY_BEAST = 30.0 FEAR_DECAY = 2.0 SAFETY_GAIN = 12.0 SAFETY_DECAY = 8.0 # Movement / combat (world units). Beasts are intentionally hard to solo and # damaging without instantly ending the village story. NPC_MOVE_STEP = 1 * TILE FLEE_STEP = 2 * TILE NPC_MELEE_REACH = ATTACK_RADIUS # one tile, shared with beasts (see mechanics.ATTACK_RADIUS) BASE_NPC_DAMAGE = 10 WEAPON_DAMAGE_BONUS = 0 # Coin economy: productive work pays a wage the citizen banks in the treasury, so # the nation grows rich through labour (gathering, building, hunting) rather than # only through raids. Wages land in inventory_coins; a transfer to the treasury # turns them into national wealth (treasury_deposit events). COINS_PER_GATHER = 1 COINS_PER_BUILD_TICK = 1 COINS_ON_BUILD_COMPLETE = 4 COINS_PER_BEAST_KILL = 6 GUARD_DAMAGE_MULTIPLIER = 1.5 WEAK_DAMAGE_MULTIPLIER = 0.3 BEAST_RETREAT_HEALTH = 20.0 BEAST_DESPAWN_TICKS = 15 # Deterministic world-system spawning. Positions and ids are random-looking but # stable for a world seed/tick pair. RESOURCE_SPAWN_INTERVAL = 7 MAX_RESOURCE_NODES = 18 BEAST_SPAWN_INTERVAL = 60 MAX_BEASTS = 3 # Resource regrowth: every node regrows +1 toward its cap on this cadence, so the # settlement is sustainable instead of starving once nodes are first depleted. RESOURCE_REGEN_INTERVALS = {"food": 8, "herbs": 12, "wood": 10, "weapon": 40} RESOURCE_REGEN_AMOUNT = 1 # How many of the nearest non-empty nodes foragers spread across (anti-clumping). SPREAD_NODE_CHOICES = 3 # Minimum surplus food before an NPC is willing to share with a needier neighbor. SHARE_FOOD_SURPLUS = 2 DEFAULT_ELECTION_INTERVAL_TICKS = 50 DEFAULT_ELECTION_DURATION_TICKS = 10 DEFAULT_CANNON_MIN_POPULATION = 6 DEFAULT_CANNON_TREASURY_COST = 20 DEFAULT_CANNON_WOOD_COST = 15 DEFAULT_CANNON_HP_TAX_PER_CITIZEN = 5 DEFAULT_CANNON_DAMAGE = 35 DEFAULT_CANNON_RADIUS = 5 DEFAULT_CANNON_COOLDOWN_TICKS = 20 # Goal -> allowed survival effect verbs (engine-side: the deterministic role # planner chooses from these sets, and role whitelists filter them per NPC). GOAL_ACTIONS: dict[str, list[str]] = { "engage_threat": ["attack", "defend", "communicate", "patrol"], "survive_threat_fight": ["attack", "defend", "communicate", "flee", "go_home"], "survive_threat_flee": ["flee", "communicate", "defend"], "help_ally": ["attack", "defend", "communicate"], "heal_self": ["heal"], "eat_food": ["consume"], "find_food": ["move_to_resource", "gather", "communicate", "steal"], "find_herbs": ["move_to_resource", "gather", "communicate"], "obey_directive": ["attack", "defend", "flee", "go_home", "move_to_resource", "gather", "communicate", "build", "patrol", "idle"], "settle": ["go_home", "consume", "communicate", "transfer"], "work": ["gather", "move_to_resource", "transfer", "communicate", "build", "patrol"], "routine_life": ["gather", "move_to_resource", "transfer", "communicate", "go_home", "build", "patrol"], "dead": [], } # Goal -> ranked LLM-facing suggestions, best first. These are public tools; # internal survival verbs such as gather/build/go_home map through use/move. GOAL_SUGGESTED_ACTIONS: dict[str, list[str]] = { "engage_threat": ["attack", "move", "speak"], "survive_threat_fight": ["attack", "move", "speak"], "survive_threat_flee": ["move", "speak"], "help_ally": ["attack", "speak"], "heal_self": ["use"], "eat_food": ["use"], "find_food": ["move", "use", "speak", "transfer"], "find_herbs": ["move", "use", "speak"], "obey_directive": ["speak", "move", "attack", "use", "transfer"], "settle": ["move", "use", "speak", "transfer"], "work": ["use", "move", "transfer", "speak"], "routine_life": ["use", "move", "transfer", "speak"], "dead": [], } PRIMITIVE_SURVIVAL_ACTIONS = frozenset( {"move", "speak", "attack", "use", "transfer", "vote"} ) # Internal effect verbs the resolver knows how to execute. Primitives are # translated to one of these in validate_survival_action; the verbs never leave # this module. VALID_SURVIVAL_VERBS = frozenset( { "gather", "consume", "heal", "attack", "steal", "communicate", "transfer", "flee", "defend", "idle", "move_to_resource", "move_to", "find_food", "find_herbs", "go_home", "patrol", "build", "craft", "assign_cannon_operator", "fire", "vote", } ) # --------------------------------------------------------------------------- # # Memory, relationships, and perception # --------------------------------------------------------------------------- # def add_episode( npc: Npc, tick: int, kind: str, actor_id: str, summary: str, *, target_id: str | None = None, object_id: str | None = None, subject_kind: str | None = None, perspective: str = "self", tags: list[str] | tuple[str, ...] | None = None, weight: float = 0.5, ) -> None: """Add a structured memory episode and update trust from generic primitives.""" normalized_tags = _dedupe_tags(tags or ()) episode = MemoryEpisode( tick=tick, kind=kind, actor_id=actor_id, target_id=target_id, object_id=object_id, subject_kind=subject_kind or _infer_subject_kind(actor_id, object_id), perspective=perspective, summary=summary, emotional_weight=_clamp(weight, 0.0, 1.0), tags=normalized_tags, ) npc.structured_memory.episodes.append(episode) _update_relationship_from_episode(npc, episode) if len(npc.structured_memory.episodes) > 20: npc.structured_memory.episodes.sort( key=lambda item: (item.emotional_weight, item.tick), reverse=True, ) npc.structured_memory.episodes = npc.structured_memory.episodes[:20] npc.structured_memory.episodes.sort(key=lambda item: item.tick) def _infer_subject_kind(actor_id: str, object_id: str | None) -> str: if actor_id.startswith("beast"): return "beast" if actor_id.startswith("npc"): return "npc" if object_id is not None: return "resource" return "world" def _dedupe_tags(tags: list[str] | tuple[str, ...]) -> list[str]: deduped: list[str] = [] for raw_tag in tags: tag = str(raw_tag).strip().lower() if tag and tag not in deduped: deduped.append(tag) return deduped def _update_relationship_from_episode(npc: Npc, episode: MemoryEpisode) -> None: if episode.subject_kind == "beast" and ( episode.kind == "attack" or "danger" in episode.tags or "beast" in episode.tags ): npc.relationships[episode.actor_id] = -1.0 return if episode.subject_kind != "npc" or episode.actor_id == npc.id: return if episode.kind == "steal": delta = -0.5 if episode.perspective == "recipient" else -0.25 _adjust_relationship(npc, episode.actor_id, delta) return if episode.kind == "attack": if "help" in episode.tags and episode.perspective in {"recipient", "witness"}: _adjust_relationship(npc, episode.actor_id, 0.3) return delta = -0.55 if episode.perspective == "recipient" else -0.25 _adjust_relationship(npc, episode.actor_id, delta) return if episode.kind in {"defend", "heal", "transfer"} and "help" in episode.tags: delta = 0.25 if episode.perspective == "recipient" else 0.1 _adjust_relationship(npc, episode.actor_id, delta) def compress_to_facts(npc: Npc) -> None: """Compress repeated episodes into simple semantic facts.""" facts: list[str] = [] by_actor: dict[str, list[MemoryEpisode]] = {} for episode in npc.structured_memory.episodes: by_actor.setdefault(episode.actor_id, []).append(episode) for actor_id, episodes in sorted(by_actor.items()): danger_events = [ episode for episode in episodes if "danger" in episode.tags or episode.kind in {"attack", "steal"} ] helpful_events = [ episode for episode in episodes if "help" in episode.tags and episode.kind in {"attack", "defend", "transfer", "heal"} ] betrayal_events = [ episode for episode in episodes if episode.kind == "steal" or "betrayal" in episode.tags ] if actor_id.startswith("beast") and danger_events: facts.append(f"{actor_id} is dangerous") if helpful_events: facts.append(f"{actor_id} is friendly") if betrayal_events: facts.append(f"{actor_id} is untrustworthy") if any(episode.kind == "attack" and episode.subject_kind == "npc" for episode in episodes): facts.append(f"{actor_id} is dangerous") deduped: list[str] = [] for fact in facts: if fact not in deduped: deduped.append(fact) npc.structured_memory.facts = deduped[:10] def build_memory_summary(npc: Npc, *, last_n: int = 3, current_tick: int | None = None) -> str: lines: list[str] = [] for fact in npc.structured_memory.facts[:10]: lines.append(f"- {fact}") recent = sorted(npc.structured_memory.episodes, key=lambda item: item.tick)[-last_n:] for episode in recent: age = "" if current_tick is not None: delta = max(0, current_tick - episode.tick) age = f" ({delta} ticks ago)" tag = _memory_tag(episode) lines.append(f"- {episode.summary}{age} [{tag}]") if not lines: legacy = [memory.text for memory in npc.memory[-last_n:]] lines.extend(f"- {text}" for text in legacy) return "\n".join(lines) if lines else "- no important memories" def build_relationships_summary(npc: Npc, world: WorldState | None = None) -> str: names = {other.id: other.name for other in world.living_npcs()} if world is not None else {} ids = set(npc.relationships) if world is not None: ids.update(other.id for other in world.living_npcs() if other.id != npc.id) ids.update(beast.id for beast in world.beasts) lines: list[str] = [] for target_id in sorted(ids): if target_id == npc.id: continue trust = -1.0 if target_id.startswith("beast") else npc.relationships.get(target_id, 0.0) label = _relationship_label(trust) display = names.get(target_id, target_id) if target_id.startswith("beast"): lines.append(f"- {display}: enemy (fear: {trust:.1f})") else: lines.append(f"- {display}: {label} (trust: {trust:.1f})") return "\n".join(lines) if lines else "- no known relationships" def build_perception(npc: Npc, world: WorldState) -> str: lines: list[str] = [] country = country_for_npc(world, npc) if country is not None: ruler = _npc_by_id(world, country.ruler_id or "") treasury = country.treasury.resources if country.treasury is not None else {} lines.append( f"- your country is {country.name} [{country.badge}], ruler=" f"{ruler.name if ruler else country.ruler_id or 'none'}" ) lines.append(f"- country policy: {country.policy}") lines.append( "- treasury " + " ".join(f"{key}={value}" for key, value in sorted(treasury.items())) ) if country.cannon is not None: operator = _npc_by_id(world, country.cannon.operator_id or "") lines.append( f"- cannon {country.cannon.id} operator=" f"{operator.name if operator else country.cannon.operator_id or 'none'}" ) election = active_election_for_country(world, country.id) if election is not None: lines.append( f"- election active until tick {election.end_tick}; candidates: " + ", ".join(election.candidate_ids) ) for rival in world.countries: if rival.id == country.id: continue rival_ruler = _npc_by_id(world, rival.ruler_id or "") rival_treasury = rival.treasury.resources if rival.treasury is not None else {} lines.append( f"- rival country {rival.name} [{rival.badge}] ruler=" f"{rival_ruler.name if rival_ruler else rival.ruler_id or 'none'}; " f"treasury {rival.treasury.id if rival.treasury else 'none'} " + " ".join(f"{key}={value}" for key, value in sorted(rival_treasury.items())) ) for beast in sorted(world.beasts, key=lambda item: vec_distance(npc.position, item.position)): if beast.state == "dead": continue distance = vec_distance(npc.position, beast.position) if distance <= PERCEPTION_BEAST_RADIUS: lines.append( f"- {beast.id} is {_tiles(distance):.0f} tiles " f"{_direction(npc.position, beast.position)} [danger]" ) for node in sorted(world.resource_nodes, key=lambda item: vec_distance(npc.position, item.position)): distance = vec_distance(npc.position, node.position) if distance <= PERCEPTION_RESOURCE_RADIUS and node.amount > 0: lines.append( f"- {node.id} is {_tiles(distance):.0f} tiles " f"{_direction(npc.position, node.position)} " f"[{node.resource_type}={node.amount}]" ) for house in sorted(world.houses, key=lambda item: vec_distance(npc.position, item.position)): distance = vec_distance(npc.position, house.position) if distance <= PERCEPTION_RESOURCE_RADIUS: lines.append( f"- {house.id} is {_tiles(distance):.0f} tiles " f"{_direction(npc.position, house.position)} " f"[{house.state} hp={house.hp:.0f} occupants={len(house.occupant_ids)}/{house.capacity}]" ) for other in sorted(npcs_in_radius(world, npc.position, PERCEPTION_NPC_RADIUS, exclude_id=npc.id), key=lambda item: item.id): trust = npc.relationships.get(other.id, 0.0) label = _relationship_label(trust) country_label = other.country_id or "no-country" inventory = [f"{label} trust={trust:.1f}", f"country={country_label}"] if other.inventory_food: inventory.append(f"food={other.inventory_food}") if other.inventory_weapon: inventory.append(f"weapon={other.inventory_weapon}") if other.hunger > STARVATION_THRESHOLD and other.inventory_food == 0: inventory.append("hungry") lines.append(f"- {other.name} nearby [{' '.join(inventory)}]") for event in world.active_events: if event.position is None or vec_distance(npc.position, event.position) <= event.radius: lines.append(f"- [EVENT] {event.description}") needs: list[str] = [] if npc.hunger > STARVATION_THRESHOLD: needs.append(f"hunger={npc.hunger:.0f}") if npc.health < FIND_HERBS_HEALTH: needs.append(f"health={npc.health:.0f}") if npc.fear > 50: needs.append(f"fear={npc.fear:.0f}") if needs: lines.append(f"- own urgent needs [{' '.join(needs)}]") return "\n".join(lines) if lines else "- nothing urgent nearby" # --------------------------------------------------------------------------- # # Spatial queries # --------------------------------------------------------------------------- # def nearest_beast(world: WorldState, position: Vec3) -> Beast | None: living = [beast for beast in world.beasts if beast.state != "dead"] if not living: return None return min(living, key=lambda beast: (vec_distance(position, beast.position), beast.id)) def nearest_beast_distance(world: WorldState, npc: Npc) -> float: beast = nearest_beast(world, npc.position) if beast is None: return float("inf") return vec_distance(npc.position, beast.position) def find_nearest_alive_npc( world: WorldState, position: Vec3, *, exclude_id: str | None = None, ) -> Npc | None: candidates = [ npc for npc in world.living_npcs() if npc.id != exclude_id ] if not candidates: return None return min(candidates, key=lambda npc: (vec_distance(position, npc.position), npc.id)) def nearest_alive_npc( world: WorldState, npc: Npc, *, max_dist: float, require_food: bool = False, ) -> Npc | None: candidates = [ other for other in world.living_npcs() if other.id != npc.id and vec_distance(npc.position, other.position) <= max_dist and (not require_food or other.inventory_food > 0) ] if not candidates: return None return min(candidates, key=lambda other: (vec_distance(npc.position, other.position), other.id)) def _nearest_role( world: WorldState, npc: Npc, role: str, *, max_dist: float, ) -> Npc | None: candidates = [ other for other in world.living_npcs() if other.id != npc.id and normalize_role(other.role) == role and vec_distance(npc.position, other.position) <= max_dist ] if not candidates: return None return min(candidates, key=lambda other: (vec_distance(npc.position, other.position), other.id)) def nearest_resource( world: WorldState, position: Vec3, *, resource_type: str | None = None, max_dist: float | None = None, require_amount: bool = True, ) -> ResourceNode | None: candidates = [ node for node in world.resource_nodes if (resource_type is None or node.resource_type == resource_type) and (not require_amount or node.amount > 0) and (max_dist is None or vec_distance(position, node.position) <= max_dist) ] if not candidates: return None return min(candidates, key=lambda node: (vec_distance(position, node.position), node.id)) def npcs_in_radius( world: WorldState, position: Vec3, radius: float, *, exclude_id: str | None = None, ) -> list[Npc]: return [ npc for npc in world.living_npcs() if npc.id != exclude_id and vec_distance(position, npc.position) <= radius ] def count_alive_npcs_in_radius(world: WorldState, position: Vec3, radius: float) -> int: return sum( 1 for npc in world.living_npcs() if vec_distance(position, npc.position) <= radius ) def count_alive_population(world: WorldState) -> int: return len(world.living_npcs()) def completed_houses(world: WorldState) -> list[House]: return [house for house in world.houses if house.state == "completed" and house.hp > 0] def house_containing_npc(world: WorldState, npc: Npc) -> House | None: for house in completed_houses(world): if vec_distance(npc.position, house.position) <= HOUSE_RADIUS: return house return None def is_npc_inside_house(world: WorldState, npc: Npc) -> bool: return house_containing_npc(world, npc) is not None def nearest_house(world: WorldState, position: Vec3, *, completed_only: bool = True) -> House | None: houses = completed_houses(world) if completed_only else [house for house in world.houses if house.hp > 0] if not houses: return None return min(houses, key=lambda house: (vec_distance(position, house.position), house.id)) def nearest_resource_for_npc( world: WorldState, npc: Npc, *, resource_type: str | None = None, max_dist: float | None = None, require_amount: bool = True, ) -> ResourceNode | None: if normalize_role(npc.role) == "builder": resource_type = "wood" return nearest_resource( world, npc.position, resource_type=resource_type, max_dist=max_dist, require_amount=require_amount, ) def country_for_npc(world: WorldState, npc: Npc) -> CountryState | None: if npc.country_id is None: return None return next((country for country in world.countries if country.id == npc.country_id), None) def active_election_for_country( world: WorldState, country_id: str, *, tick: int | None = None, ) -> ElectionState | None: current_tick = world.tick if tick is None else tick for election in world.elections: if election.country_id != country_id or election.completed: continue if election.start_tick <= current_tick <= election.end_tick: return election return None def active_election_for_npc(world: WorldState, npc: Npc, *, tick: int | None = None) -> ElectionState | None: if npc.country_id is None: return None return active_election_for_country(world, npc.country_id, tick=tick) def _env_int(name: str, default: int) -> int: raw = os.getenv(name) if raw is None: return default try: value = int(raw) except ValueError: return default return value if value > 0 else default def apply_world_systems(world: WorldState) -> None: """Run deterministic environment systems before NPC decisions.""" # Local import: chaos.py reuses survival helpers at module level. from world_simulator.simulation.chaos import end_famine_if_due end_famine_if_due(world) if world.resource_nodes: _regrow_resource_nodes(world) _maybe_spawn_resource_node(world) if world.beasts or world.resource_nodes: _maybe_spawn_beast(world) def _regrow_resource_nodes(world: WorldState) -> None: for node in world.resource_nodes: interval = RESOURCE_REGEN_INTERVALS.get(node.resource_type, 10) if world.tick % interval == 0 and node.max_amount > 0 and node.amount < node.max_amount: node.amount = min(node.max_amount, node.amount + RESOURCE_REGEN_AMOUNT) def _maybe_spawn_resource_node(world: WorldState) -> None: if world.tick == 0 or world.tick % RESOURCE_SPAWN_INTERVAL != 0: return if len(world.resource_nodes) >= MAX_RESOURCE_NODES: return rng = random.Random(f"{world.seed}:{world.tick}:resource_spawn") resource_type = rng.choices( ["food", "herbs", "wood", "weapon"], weights=[6, 3, 3, 1], k=1, )[0] max_amount = {"food": 5, "herbs": 4, "wood": 6, "weapon": 2}[resource_type] node_index = _next_resource_index(world, resource_type) world.resource_nodes.append( ResourceNode( id=f"res_{resource_type}_{node_index}", resource_type=resource_type, position=_random_map_position(world, rng), amount=max(1, rng.randint(max(1, max_amount // 2), max_amount)), max_amount=max_amount, ) ) def _maybe_spawn_beast(world: WorldState) -> None: living_beasts = [beast for beast in world.beasts if beast.state != "dead"] if len(living_beasts) >= MAX_BEASTS: return if count_alive_population(world) < 4: return cooldown = _natural_beast_spawn_interval(world) if world.tick == 0 or world.tick % cooldown != 0: return rng = random.Random(f"{world.seed}:{world.tick}:beast_spawn") beast = Beast( id=f"beast_{_next_beast_index(world)}", position=_random_map_edge_position(world, rng), health=60.0, damage=9.0, ) world.beasts.append(beast) _log_event( world, "beast_spawned", f"{beast.id} appeared at the edge of the map", severity="warning", actor_id=beast.id, ) def _natural_beast_spawn_interval(world: WorldState) -> int: rng = random.Random(f"{world.seed}:beast_spawn_interval:{world.tick // BEAST_SPAWN_INTERVAL}") return rng.randint(60, 90) def _random_map_edge_position(world: WorldState, rng: random.Random) -> Vec3: half_width = world.terrain.width / 2 half_depth = world.terrain.depth / 2 side = rng.choice(["north", "south", "east", "west"]) if side == "north": return Vec3(x=round(rng.uniform(-half_width, half_width), 3), y=0.0, z=half_depth) if side == "south": return Vec3(x=round(rng.uniform(-half_width, half_width), 3), y=0.0, z=-half_depth) if side == "east": return Vec3(x=half_width, y=0.0, z=round(rng.uniform(-half_depth, half_depth), 3)) return Vec3(x=-half_width, y=0.0, z=round(rng.uniform(-half_depth, half_depth), 3)) def _next_resource_index(world: WorldState, resource_type: str) -> int: prefix = f"res_{resource_type}_" highest = 0 for node in world.resource_nodes: if not node.id.startswith(prefix): continue suffix = node.id.removeprefix(prefix) if suffix.isdigit(): highest = max(highest, int(suffix)) return highest + 1 def _next_beast_index(world: WorldState) -> int: highest = 0 for beast in world.beasts: suffix = beast.id.removeprefix("beast_") if suffix.isdigit(): highest = max(highest, int(suffix)) return highest + 1 def _random_map_position(world: WorldState, rng: random.Random) -> Vec3: half_width = world.terrain.width / 2 half_depth = world.terrain.depth / 2 return Vec3( x=round(rng.uniform(-half_width + TILE, half_width - TILE), 3), y=0.0, z=round(rng.uniform(-half_depth + TILE, half_depth - TILE), 3), ) def _select_beast_target(world: WorldState, beast: Beast) -> Npc | None: candidates = [ npc for npc in world.living_npcs() if not is_npc_inside_house(world, npc) ] if not candidates: return None nearest_distance = min(vec_distance(beast.position, npc.position) for npc in candidates) vulnerable = [ npc for npc in candidates if npc.health < 50 and vec_distance(beast.position, npc.position) <= nearest_distance * 1.3 ] pool = vulnerable or candidates return min(pool, key=lambda npc: (vec_distance(beast.position, npc.position), npc.health, npc.id)) def _select_house_target(world: WorldState, beast: Beast) -> House | None: houses = completed_houses(world) if not houses: return None occupied_ids = set() for npc in world.living_npcs(): house = house_containing_npc(world, npc) if house is not None: occupied_ids.add(house.id) occupied = [house for house in houses if house.id in occupied_ids] pool = occupied or houses return min(pool, key=lambda house: (vec_distance(beast.position, house.position), house.id)) def _record_world_event( world: WorldState, kind: str, description: str, position: Vec3 | None, *, radius: float, duration_ticks: int, ) -> None: world.active_events.append( WorldEvent( tick_created=world.tick, kind=kind, description=description, position=position, radius=radius, duration_ticks=duration_ticks, ) ) def _log_event( world: WorldState, event_type: str, summary: str, *, severity: str = "neutral", actor_id: str | None = None, target_id: str | None = None, object_id: str | None = None, tick: int | None = None, ) -> None: normalized_severity = severity if severity in {"good", "neutral", "warning", "danger"} else "neutral" world.event_log.append( WorldLogEvent( tick=world.tick if tick is None else tick, type=event_type, actor_id=actor_id, target_id=target_id, object_id=object_id, summary=summary, severity=normalized_severity, # type: ignore[arg-type] ) ) # --------------------------------------------------------------------------- # # Deterministic environment tick (no LLM) # --------------------------------------------------------------------------- # def apply_survival_tick(world: WorldState) -> None: """Advance hunger, the beast, fear, and starvation deaths for one tick. No-op for worlds without survival entities, so the existing social-only pipeline and its tests are unaffected. """ if world.active_events: world.active_events = [ event for event in world.active_events if world.tick - event.tick_created < event.duration_ticks ] apply_world_systems(world) if not world.beasts and not world.resource_nodes and not world.houses: return _refresh_house_occupants(world) for npc in list(world.living_npcs()): npc.age += 1 npc.reproduction_cooldown = max(0, npc.reproduction_cooldown - 1) npc.hunger = min(HUNGER_CAP, npc.hunger + HUNGER_RATE) if npc.hunger >= SEVERE_STARVATION_THRESHOLD: npc.health -= SEVERE_STARVATION_DAMAGE elif npc.hunger >= STARVATION_THRESHOLD: npc.health -= STARVATION_DAMAGE if npc.health <= 0: _kill_npc(world, npc, cause="starvation", actor_id=npc.id) continue if npc.age >= npc.max_age: _kill_npc(world, npc, cause="old_age", actor_id=npc.id) _apply_beast_system(world) _refresh_house_occupants(world) for npc in world.living_npcs(): _update_safety(world, npc) if nearest_beast_distance(world, npc) > FEAR_DECAY_RADIUS: npc.fear = max(0.0, npc.fear - FEAR_DECAY) # Keep the live goal current for the snapshot, even when planning ran on a # deep-copied world (the HTTP server path). npc.survival_goal = select_goal(npc, world) if world.tick > 0 and world.tick % 10 == 0: compress_to_facts(npc) def apply_post_survival_tick(world: WorldState, next_tick: int) -> None: """Run deterministic end-of-tick systems after NPC actions resolve.""" _refresh_house_occupants(world) _refresh_country_citizens(world, next_tick) _run_election_system(world, next_tick) for npc in list(world.npcs): if is_alive(npc): _maybe_reproduce(world, npc, next_tick) _refresh_house_occupants(world) _refresh_country_citizens(world, next_tick) if next_tick % 10 == 0: _recompute_importance(world) _update_population_and_status(world, next_tick) def _refresh_country_citizens(world: WorldState, next_tick: int) -> None: living_ids = {npc.id for npc in world.living_npcs()} for country in world.countries: country.citizen_ids = [ npc.id for npc in world.npcs if npc.country_id == country.id and npc.id in living_ids ] if country.ruler_id not in living_ids or country.ruler_id not in country.citizen_ids: country.ruler_id = country.citizen_ids[0] if country.citizen_ids else None for npc in world.npcs: if npc.country_id != country.id: continue npc.special_status = "ruler" if npc.id == country.ruler_id else ( "cannon_operator" if country.cannon and country.cannon.operator_id == npc.id else None ) def _run_election_system(world: WorldState, next_tick: int) -> None: for country in world.countries: active = active_election_for_country(world, country.id, tick=next_tick) if active is None and country.citizen_ids and next_tick >= country.next_election_tick: _start_election(world, country, next_tick) active = active_election_for_country(world, country.id, tick=next_tick) if active is not None and next_tick >= active.end_tick: _finish_election(world, country, active, next_tick) def _start_election(world: WorldState, country: CountryState, tick: int) -> None: duration = _env_int("ELECTION_DURATION_TICKS", DEFAULT_ELECTION_DURATION_TICKS) candidate_ids = list(country.citizen_ids) election = ElectionState( country_id=country.id, start_tick=tick, end_tick=tick + duration, candidate_ids=candidate_ids, ) world.elections.append(election) _log_event( world, "election_started", f"{country.name} election started with {len(candidate_ids)} candidates", severity="neutral", object_id=country.id, tick=tick, ) def _finish_election( world: WorldState, country: CountryState, election: ElectionState, tick: int, ) -> None: for npc_id in country.citizen_ids: if npc_id in election.votes: continue election.votes[npc_id] = _fallback_vote(world, election, voter_id=npc_id) counts: dict[str, int] = {} for candidate_id in election.votes.values(): if candidate_id in election.candidate_ids: counts[candidate_id] = counts.get(candidate_id, 0) + 1 if counts: winner_id = min(counts, key=lambda candidate_id: (-counts[candidate_id], candidate_id)) country.ruler_id = winner_id _set_country_policy(world, country, tick) election.completed = True country.next_election_tick = tick + _env_int( "ELECTION_INTERVAL_TICKS", DEFAULT_ELECTION_INTERVAL_TICKS, ) _refresh_country_citizens(world, tick) _log_event( world, "election_completed", f"{country.name} elected {country.ruler_id} with {len(election.votes)} votes", severity="good", target_id=country.ruler_id, object_id=country.id, tick=tick, ) def _fallback_vote(world: WorldState, election: ElectionState, *, voter_id: str) -> str: if not election.candidate_ids: return voter_id rng = random.Random(f"{world.seed}:{election.country_id}:{election.start_tick}:{voter_id}") return rng.choice(election.candidate_ids) def _set_country_policy(world: WorldState, country: CountryState, tick: int) -> None: treasury = country.treasury.resources if country.treasury else {} wood = treasury.get("wood", 0) coins = treasury.get("coins", 0) if wood >= DEFAULT_CANNON_WOOD_COST and coins >= DEFAULT_CANNON_TREASURY_COST: country.policy = "Craft a cannon, protect the treasury, and attack only clear threats." elif coins < 8: country.policy = "Gather coins and wood, defend citizens, and avoid reckless fights." else: country.policy = "Gather resources, help weak citizens, and watch the other country." _log_event( world, "policy_updated", f"{country.name} policy: {country.policy}", severity="neutral", actor_id=country.ruler_id, object_id=country.id, tick=tick, ) def _apply_beast_system(world: WorldState) -> None: half_width = world.terrain.width / 2 half_depth = world.terrain.depth / 2 for beast in world.beasts: if beast.state == "dead": continue target = _select_beast_target(world, beast) house_target = None if target is not None else _select_house_target(world, beast) beast.target_npc_id = target.id if target is not None else None beast.target_house_id = house_target.id if house_target is not None else None target_position = target.position if target is not None else ( house_target.position if house_target is not None else None ) if target_position is None: continue step = beast.speed * TILE reach = ATTACK_RADIUS if beast.health < BEAST_RETREAT_HEALTH: beast.state = "retreating" beast.retreat_ticks += 1 beast.position = move_away( beast.position, target_position, step, half_width=half_width, half_depth=half_depth, ) if beast.retreat_ticks >= BEAST_DESPAWN_TICKS: beast.state = "dead" _log_event( world, "beast_retreat", f"{beast.id} fled the map", severity="neutral", actor_id=beast.id, ) continue beast.retreat_ticks = 0 distance = vec_distance(beast.position, target_position) if distance > reach: beast.state = "hunting" beast.position = move_toward( beast.position, target_position, step, half_width=half_width, half_depth=half_depth, ) continue beast.state = "attacking" if target is not None: _beast_attack_npc(world, beast, target) elif house_target is not None: _damage_house(world, house_target, beast, amount=5.0) def _beast_attack_npc(world: WorldState, beast: Beast, target: Npc) -> None: if is_npc_inside_house(world, target): house = house_containing_npc(world, target) if house is not None: _damage_house(world, house, beast, amount=5.0) return target.health -= beast.damage target.fear = min(100.0, max(target.fear, FEAR_ON_ATTACK)) add_episode( target, world.tick, "attack", beast.id, f"{beast.id} attacked me", target_id=target.id, subject_kind="beast", perspective="recipient", tags=["danger", "beast"], weight=1.0, ) _record_world_event( world, "beast_attack", f"{beast.id} attacked {target.name}", beast.position, radius=WITNESS_RADIUS, duration_ticks=4, ) _log_event( world, "beast_attack", f"{beast.id} attacked {target.name}", severity="danger", actor_id=beast.id, target_id=target.id, ) remember(target, world.tick, f"Beast {beast.id} attacked me at tick {world.tick}.") if target.health <= 0: _kill_npc(world, target, cause="beast", actor_id=beast.id) for witness in npcs_in_radius(world, beast.position, WITNESS_RADIUS, exclude_id=target.id): witness.fear = min(100.0, witness.fear + FEAR_ON_WITNESS) add_episode( witness, world.tick, "attack", beast.id, f"{beast.id} attacked {target.name} nearby", target_id=target.id, subject_kind="beast", perspective="witness", tags=["danger", "beast"], weight=0.8, ) remember(witness, world.tick, f"The beast attacked {target.name} nearby.") def _damage_house(world: WorldState, house: House, beast: Beast, *, amount: float) -> None: house.hp = max(0.0, house.hp - amount) _record_world_event( world, "house_damaged", f"{beast.id} damaged {house.id}", house.position, radius=WITNESS_RADIUS, duration_ticks=4, ) _log_event( world, "house_damaged", f"{beast.id} damaged {house.id}", severity="warning", actor_id=beast.id, target_id=house.id, ) if house.hp > 0: return house.state = "destroyed" house.build_progress = 0 house.occupant_ids = [] if _chaos_points_active(world): world.chaos_score += 2 _record_world_event( world, "house_destroyed", f"{house.id} was destroyed", house.position, radius=WITNESS_RADIUS, duration_ticks=6, ) _log_event( world, "house_destroyed", f"{house.id} was destroyed", severity="danger", actor_id=beast.id, target_id=house.id, ) for npc in npcs_in_radius(world, house.position, HOUSE_RADIUS): npc.fear = max(npc.fear, 80.0) npc.safety = 0.0 remember(npc, world.tick, f"{house.id} was destroyed around you.") def _kill_npc(world: WorldState, npc: Npc, *, cause: str, actor_id: str | None) -> None: if npc.health <= 0 and npc.intention == "dead": return npc.health = min(npc.health, 0) npc.intention = "dead" npc.survival_goal = "dead" clear_active_directive(npc) world.deaths_by_cause[cause] = world.deaths_by_cause.get(cause, 0) + 1 if _chaos_points_active(world): world.chaos_score += 3 add_episode( npc, world.tick, "death", actor_id or npc.id, f"{npc.name} died of {cause}", target_id=npc.id, subject_kind="beast" if actor_id and actor_id.startswith("beast") else "npc", perspective="recipient", tags=["danger", "death", cause], weight=1.0, ) remember(npc, world.tick, "You died.") _drop_inventory(world, npc) _record_world_event( world, "npc_died", f"{npc.name} died of {cause}", npc.position, radius=WITNESS_RADIUS, duration_ticks=6, ) _log_event( world, "npc_died", f"{npc.name} died of {cause}", severity="danger", actor_id=actor_id, target_id=npc.id, object_id=cause, ) for witness in npcs_in_radius(world, npc.position, WITNESS_RADIUS, exclude_id=npc.id): actor_is_beast = bool(actor_id and actor_id.startswith("beast")) actor_is_npc = bool(actor_id and actor_id.startswith("npc")) if actor_is_beast: witness.fear = min(100.0, witness.fear + FEAR_ON_DEATH_BY_BEAST) witness.relationships[actor_id or "beast"] = -1.0 else: witness.fear = min(100.0, witness.fear + FEAR_ON_WITNESS) if actor_is_npc and actor_id is not None: _adjust_relationship(witness, actor_id, -0.5) add_episode( witness, world.tick, "death", actor_id or npc.id, f"{npc.name} died of {cause}", target_id=npc.id, subject_kind="beast" if actor_is_beast else "npc", perspective="witness", tags=["danger", "death", cause], weight=0.9, ) remember(witness, world.tick, f"{npc.name} died of {cause}.") _purge_dead_npc_references(world, npc) def _drop_inventory(world: WorldState, npc: Npc) -> None: for resource_type, amount in ( ("food", npc.inventory_food), ("herbs", npc.inventory_herbs), ("wood", npc.inventory_wood), ("weapon", npc.inventory_weapon), ("coins", npc.inventory_coins), ): if amount <= 0: continue index = _next_resource_index(world, resource_type) world.resource_nodes.append( ResourceNode( id=f"res_{resource_type}_{index}", resource_type=resource_type, position=npc.position, amount=amount, max_amount=max( amount, {"food": 5, "herbs": 4, "wood": 6, "weapon": 2, "coins": 10}[resource_type], ), ) ) npc.inventory_food = 0 npc.inventory_herbs = 0 npc.inventory_wood = 0 npc.inventory_weapon = 0 npc.inventory_coins = 0 def _purge_dead_npc_references(world: WorldState, dead: Npc) -> None: for house in world.houses: house.occupant_ids = [npc_id for npc_id in house.occupant_ids if npc_id != dead.id] house.owner_ids = [npc_id for npc_id in house.owner_ids if npc_id != dead.id] dead.focus_target_id = None dead.help_target_id = None dead.build_target_house_id = None dead.home_house_id = None dead_name = dead.name.lower() dead_id = dead.id.lower() for npc in world.living_npcs(): if npc.focus_target_id == dead.id: npc.focus_target_id = None if npc.help_target_id == dead.id: npc.help_target_id = None if npc.god_directive: lowered = npc.god_directive.lower() if dead_id in lowered or dead_name in lowered: clear_active_directive(npc) for beast in world.beasts: if beast.target_npc_id == dead.id: beast.target_npc_id = None def _refresh_house_occupants(world: WorldState) -> None: for house in world.houses: if house.state != "completed" or house.hp <= 0: house.occupant_ids = [] continue house.occupant_ids = [ npc.id for npc in world.living_npcs() if vec_distance(npc.position, house.position) <= HOUSE_RADIUS ][: house.capacity] def _update_safety(world: WorldState, npc: Npc) -> None: house = house_containing_npc(world, npc) if house is None: npc.safety = max(0.0, npc.safety - SAFETY_DECAY) npc.needs.safety = round(npc.safety) return threat_near_house = any( beast.state != "dead" and vec_distance(beast.position, house.position) <= HOUSE_THREAT_RADIUS for beast in world.beasts ) or any( other.id != npc.id and vec_distance(other.position, npc.position) <= 10 * TILE and _has_hostile_directive_toward(other, npc) for other in world.living_npcs() ) if not threat_near_house: npc.safety = min(100.0, npc.safety + SAFETY_GAIN) npc.needs.safety = round(npc.safety) def _has_hostile_directive_toward(actor: Npc, target: Npc) -> bool: if not actor.god_directive: return False lowered = actor.god_directive.lower() return "attack" in lowered or "kill" in lowered or target.name.lower() in lowered def _maybe_reproduce(world: WorldState, parent: Npc, next_tick: int) -> None: house = house_containing_npc(world, parent) if house is None: return if not _reproduction_conditions_met(world, parent): return role = _role_for_child(world, parent) child_index = _next_npc_index(world) # A child belongs to its parents' nation and is driven by the same model # (connector). Without inheriting these the newborn is stateless: no country, # no treasury to fund, and wrongly routed to the default model. child = Npc( id=f"npc-{child_index:03d}", name=_child_name(child_index), role=role, position=house.position, health=70, hunger=20.0, fear=0.0, safety=100.0 if not _house_has_threat(world, house) else 0.0, age=0, max_age=random.Random(f"{world.seed}:{next_tick}:{child_index}:max_age").randint(320, 480), inventory_weapon=0, home_house_id=house.id, country_id=parent.country_id, connector_id=parent.connector_id, unrestricted_actions=parent.unrestricted_actions, model_profile_id=parent.model_profile_id, personality=parent.personality, memory=[MemoryEntry(tick=next_tick, text=f"Born in {house.id} as a {role}.")], ) if parent.country_id is not None: country = country_for_npc(world, parent) if country is not None and child.id not in country.citizen_ids: country.citizen_ids.append(child.id) for npc in world.living_npcs(): if npc.id == child.id: continue child.relationships[npc.id] = 0.45 npc.relationships[child.id] = 0.45 child.relationships[parent.id] = 0.6 parent.relationships[child.id] = 0.6 world.npcs.append(child) parent.children_count += 1 parent.reproduction_cooldown = 100 parent.hunger = min(HUNGER_CAP, parent.hunger + 40.0) world.total_births += 1 world.overseer_score += 3 add_episode( parent, next_tick, "npc_born", parent.id, f"I had a child: {child.name}", target_id=child.id, subject_kind="npc", perspective="self", tags=["birth", "family"], weight=0.8, ) for witness in npcs_in_radius(world, house.position, HOUSE_RADIUS, exclude_id=parent.id): add_episode( witness, next_tick, "npc_born", parent.id, f"{parent.name} had a child: {child.name}", target_id=child.id, subject_kind="npc", perspective="witness", tags=["birth", "family"], weight=0.6, ) _log_event( world, "npc_born", f"{parent.name} gave birth to {child.name}", severity="good", actor_id=parent.id, target_id=child.id, tick=next_tick, ) def _reproduction_conditions_met(world: WorldState, npc: Npc) -> bool: return ( npc.health >= 95 and npc.hunger <= REPRODUCTION_MAX_HUNGER and round(npc.safety) >= 100 and npc.age > 90 and npc.reproduction_cooldown == 0 and count_alive_population(world) < world.population_cap ) def _house_has_threat(world: WorldState, house: House) -> bool: return any( beast.state != "dead" and vec_distance(beast.position, house.position) <= HOUSE_THREAT_RADIUS for beast in world.beasts ) def _role_for_child(world: WorldState, parent: Npc) -> str: # Everyone is a free citizen with no fixed job, so children are too. return "citizen" def _next_npc_index(world: WorldState) -> int: highest = 0 for npc in world.npcs: suffix = npc.id.removeprefix("npc-") if suffix.isdigit(): highest = max(highest, int(suffix)) return highest + 1 def _child_name(index: int) -> str: names = [ "Ada", "Boris", "Cora", "Dima", "Elena", "Farid", "Gita", "Hana", "Ivan", "Juno", "Kira", "Lena", "Mira", "Niko", "Oleg", "Pavel", "Raya", "Sofia", "Toma", "Vera", ] return names[(index - 1) % len(names)] def _recompute_importance(world: WorldState) -> None: counts: dict[str, int] = {} for npc in world.living_npcs(): role = normalize_role(npc.role) counts[role] = counts.get(role, 0) + 1 for npc in world.npcs: role_count = max(1, counts.get(normalize_role(npc.role), 1)) role_scarcity = 1 / role_count npc.importance = round( 1.0 + 2.0 * role_scarcity + 0.5 * npc.beasts_killed + 0.3 * npc.resources_transferred + 0.4 * npc.children_count + 0.5 * npc.houses_built, 3, ) def _update_population_and_status(world: WorldState, next_tick: int) -> None: population = count_alive_population(world) world.population = population world.peak_population = max(world.peak_population, population) if world.game_status != "running": return if population <= 0: world.game_status = "lose" elif population >= world.population_cap: world.game_status = "win_population" elif next_tick >= 600: world.game_status = "win_survival" if world.game_status != "running": _log_event( world, "game_over", _game_over_summary(world), severity="good" if world.game_status.startswith("win") else "danger", tick=next_tick, ) def _game_over_summary(world: WorldState) -> str: return ( f"{world.game_status}: births={world.total_births}, deaths={sum(world.deaths_by_cause.values())}, " f"houses_built={world.houses_built}, beasts_killed={world.beasts_killed}, " f"peak_population={world.peak_population}, score={world.overseer_score}-{world.chaos_score}" ) def _chaos_points_active(world: WorldState) -> bool: return world.tick <= world.chaos_intervention_until # --------------------------------------------------------------------------- # # Goal selection + action mask # --------------------------------------------------------------------------- # def select_goal(npc: Npc, world: WorldState) -> str: if not is_alive(npc): return "dead" role = normalize_role(npc.role) if npc.help_target_id: target = _npc_by_id(world, npc.help_target_id) if ( target is not None and is_alive(target) and role == "guard" and npc.relationships.get(target.id, 0.0) > 0.3 and nearest_beast_distance(world, target) < BEAST_ALERT_RADIUS ): return "help_ally" npc.help_target_id = None # NOTE: a nearby beast no longer forces a flee/engage goal. The beast is shown # in the briefing's THREATS section (with attack/move options) and the LLM # decides for itself whether to fight, flee, or call for help. if npc.health < HEAL_GOAL_HEALTH and npc.inventory_herbs > 0: return "heal_self" if ( _is_reproduction_candidate(npc) and npc.inventory_food > 0 and REPRODUCTION_MAX_HUNGER < npc.hunger <= REPRODUCTION_SNACK_HUNGER ): return "eat_food" if npc.hunger >= EAT_GOAL_HUNGER and npc.inventory_food > 0: return "eat_food" if npc.god_directive: return "obey_directive" if npc.hunger >= EAT_GOAL_HUNGER: return "find_food" if npc.health < FIND_HERBS_HEALTH and npc.inventory_herbs == 0: return "find_herbs" if npc.health >= 80 and npc.hunger <= 30 and not _has_any_living_beast_near(world, npc.position, HOUSE_THREAT_RADIUS): return "settle" return "work" def suggested_actions_for_goal(goal: str) -> list[str]: return list(GOAL_SUGGESTED_ACTIONS.get(goal, [])) def allowed_actions_for_goal(goal: str) -> list[str]: return list(GOAL_ACTIONS.get(goal, [])) def allowed_actions_for_npc(world: WorldState, npc: Npc, goal: str) -> list[str]: allowed = allowed_actions_for_goal(goal) if goal == "survive_threat_flee": allowed = [action for action in allowed if action != "transfer"] if "gather" in allowed and not _has_gatherable_resource_for_goal(world, npc, goal): allowed = [action for action in allowed if action != "gather"] if "consume" in allowed and npc.inventory_food <= 0: allowed = [action for action in allowed if action != "consume"] if "heal" in allowed and npc.inventory_herbs <= 0: allowed = [action for action in allowed if action != "heal"] return allowed or ["idle"] def _has_gatherable_resource_for_goal(world: WorldState, npc: Npc, goal: str) -> bool: resource_type: str | None = None if goal == "find_food": resource_type = "food" elif goal == "find_herbs": resource_type = "herbs" elif normalize_role(npc.role) == "builder": resource_type = "wood" node = nearest_resource_for_npc(world, npc, resource_type=resource_type) return node is not None and vec_distance(npc.position, node.position) <= GATHER_RADIUS def suggested_actions_for_npc(npc: Npc, goal: str) -> list[str]: """Ranked LLM suggestions. Roles influence planning context, not tool access.""" suggested = suggested_actions_for_goal(goal) return suggested or ["move", "speak"] def _beast_threatens_village(world: WorldState) -> bool: for beast in world.beasts: if beast.state == "dead": continue for house in completed_houses(world): if vec_distance(beast.position, house.position) <= VILLAGE_THREAT_RADIUS: return True for npc in world.living_npcs(): if vec_distance(beast.position, npc.position) <= VILLAGE_THREAT_RADIUS: return True return False def _has_any_living_beast_near(world: WorldState, position: Vec3, radius: float) -> bool: return any( beast.state != "dead" and vec_distance(beast.position, position) <= radius for beast in world.beasts ) def _is_reproduction_candidate(npc: Npc) -> bool: return ( npc.health >= 95 and npc.age > 90 and npc.reproduction_cooldown == 0 and round(npc.safety) >= 100 ) def validate_survival_action( npc: Npc, action: str, world: WorldState, directive: NpcDirective | None = None, ) -> str: """Translate a requested action into a safe executable effect verb. Primitives are mapped onto effect verbs first; then role whitelists and physical impossibility (missing nodes, empty inventory, no target) are repaired so the engine never executes an illegal request. """ action = _canonical_action(_verb_for_action(npc, action, world, directive)) if action not in VALID_SURVIVAL_VERBS: return "idle" resource_type = _resource_type_for_validation(world, npc, directive, action) if action == "gather": node = _resource_from_directive(world, directive, npc, max_dist=GATHER_RADIUS) if node is None or node.amount <= 0: return "move_to_resource" return "gather" if action == "consume": return "consume" if npc.inventory_food > 0 else "find_food" if action == "heal": return "heal" if npc.inventory_herbs > 0 else "find_herbs" if action == "attack": if _melee_target_npc(world, npc, directive) is not None: return "attack" # explicit NPC target; effect approaches if needed beast = _beast_from_directive(world, npc, directive) if beast is None: return "defend" return "attack" # apply_action_effects approaches when out of melee reach if action == "build": if npc.build_target_house_id: house = _house_by_id(world, npc.build_target_house_id) if house is not None and house.state == "under_construction": return "build" if ( _nearest_damaged_house(world, npc.position) is not None and npc.inventory_wood >= HOUSE_REPAIR_WOOD_COST ): return "build" return "build" if npc.inventory_wood >= 5 else "move_to_resource" if action == "craft": return "craft" if _can_craft_cannon(world, npc) else "idle" if action == "assign_cannon_operator": return "assign_cannon_operator" if _can_assign_cannon_operator(world, npc, directive) else "idle" if action == "fire": return "fire" if _can_fire_cannon(world, npc, directive) else "idle" if action == "go_home": return "go_home" if nearest_house(world, npc.position) is not None else "move_to" if action == "patrol": return "patrol" if action == "steal": partner = _target_npc_from_directive(world, npc, directive, max_dist=TRADE_RADIUS) if partner is None: partner = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS, require_food=True) return "steal" if partner is not None else "move_to_resource" if action == "transfer": if _treasury_from_directive(world, directive) is not None: return "transfer" partner = _target_npc_from_directive(world, npc, directive, max_dist=TRADE_RADIUS) if partner is None: partner = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS) return "transfer" if partner is not None else "communicate" if action == "vote": election = active_election_for_npc(world, npc) candidate_id = directive.target_npc_id if directive is not None else None if election is None or candidate_id not in election.candidate_ids: return "idle" return "vote" return action def _verb_for_action( npc: Npc, action: str, world: WorldState, directive: NpcDirective | None, ) -> str: """Map a primitive plus its parameters onto an internal effect verb. Non-primitive verbs (already translated, or internal repairs) pass through. """ if action not in PRIMITIVE_SURVIVAL_ACTIONS: return action if action == "move": if directive is not None and directive.away: return "flee" if directive is not None and ( directive.resource_id or directive.resource_type or directive.target ): return "move_to_resource" return "move_to" if action == "speak": return "communicate" if action in {"attack", "strike"}: return "attack" if action == "use": use_type = (directive.use_type or "").lower() if directive is not None else "" if use_type in {"eat", "consume"}: return "consume" if use_type == "heal": return "heal" if use_type == "gather": return "gather" if use_type in {"build", "repair"}: return "build" if use_type == "craft": return "craft" if use_type == "assign_cannon_operator": return "assign_cannon_operator" if use_type == "fire": return "fire" if directive is not None and directive.resource_id: return "gather" resource_type = directive.resource_type if directive is not None else None if resource_type == "food": return "consume" if resource_type == "herbs": return "heal" node = _resource_from_directive(world, directive, npc, max_dist=GATHER_RADIUS) if node is not None and node.amount > 0: return "gather" if npc.inventory_food > 0 and npc.hunger > STARVATION_THRESHOLD: return "consume" return "gather" if action == "transfer": # A transfer aimed at a treasury always resolves through the treasury # handler, which honors `take` (steal from a rival) vs deposit. Routing # take->NPC-steal here would bypass treasury raids entirely. if _treasury_from_directive(world, directive) is not None: return "transfer" if directive is not None and directive.take: return "steal" return "transfer" if action == "vote": return "vote" return "idle" def _canonical_action(action: str) -> str: return canonical_action(action) def _resource_type_for_validation( world: WorldState, npc: Npc, directive: NpcDirective | None, action: str, ) -> str | None: if directive is not None and directive.resource_type: return directive.resource_type if directive is not None and directive.resource_id: node = next((node for node in world.resource_nodes if node.id == directive.resource_id), None) return node.resource_type if node is not None else None if action in {"gather", "move_to_resource"}: node = nearest_resource_for_npc(world, npc) return node.resource_type if node is not None else None return None def _role_fallback_action(world: WorldState, npc: Npc) -> str: role = normalize_role(npc.role) if role == "guard": if nearest_beast(world, npc.position) is not None: return "attack" return "patrol" if role == "builder": if npc.inventory_wood >= 5: return "build" return "move_to_resource" if npc.inventory_food > 0 and npc.hunger >= EAT_GOAL_HUNGER: return "consume" return "move_to_resource" def _border_block_summary( dest: Vec3, half_width: float, half_depth: float, ) -> str | None: """When the NPC aimed PAST the world border (an off-map, impassable target), return a short message it sees as LAST TURN feedback so it learns the move was blocked at the edge. None for normal in-bounds targets. Convention (matches the briefing compass): +X=WEST, -X=EAST, +Z=NORTH, -Z=SOUTH.""" eps = 1e-3 blocked: list[str] = [] if dest.x > half_width + eps: blocked.append("WEST") elif dest.x < -half_width - eps: blocked.append("EAST") if dest.z > half_depth + eps: blocked.append("NORTH") elif dest.z < -half_depth - eps: blocked.append("SOUTH") if not blocked: return None edge = " and ".join(blocked) return ( f"you hit the world border to the {edge}: your target X={dest.x:g} Z={dest.z:g} " f"is OFF THE MAP (it ends at X +-{half_width:g}, Z +-{half_depth:g}). You " "can go no further that way - aim for a target INSIDE the map" ) # --------------------------------------------------------------------------- # # Effect resolver (engine owns all state changes) # --------------------------------------------------------------------------- # def apply_action_effects( npc: Npc, action: str, world: WorldState, *, destination: Vec3 | None = None, directive: NpcDirective | None = None, ) -> str: """Apply one validated survival action; return a short intention summary.""" action = _verb_for_action(npc, action, world, directive) npc.focus_target_id = None half_width = world.terrain.width / 2 half_depth = world.terrain.depth / 2 if action == "consume": if npc.inventory_food > 0: npc.inventory_food -= 1 npc.hunger = max(0.0, npc.hunger - 40.0) npc.health = min(100, npc.health + 5) add_episode( npc, world.tick, "consume", npc.id, "I ate food", object_id="food", subject_kind="npc", perspective="self", tags=["food"], weight=0.2, ) _log_event( world, "consume", f"{npc.name} ate food", actor_id=npc.id, object_id="food", ) return "eating food" return "searching for food" if action == "heal": if npc.inventory_herbs > 0: npc.inventory_herbs -= 1 npc.health = min(100, npc.health + 25) _log_event( world, "heal", f"{npc.name} healed with herbs", severity="good", actor_id=npc.id, object_id="herbs", ) return "healing with herbs" return "searching for herbs" if action == "gather": node = _resource_from_directive(world, directive, npc, max_dist=GATHER_RADIUS) if node is not None and node.amount > 0: node.amount -= 1 _add_to_inventory(npc, node.resource_type) _earn_coins(npc, world, COINS_PER_GATHER, f"gathering {node.resource_type}") add_episode( npc, world.tick, "gather", npc.id, f"I gathered {node.resource_type} from {node.id}", object_id=node.id, subject_kind="resource", perspective="self", tags=[node.resource_type], weight=0.3, ) remember(npc, world.tick, f"You gathered {node.resource_type}.") _log_event( world, "gather", f"{npc.name} gathered {node.resource_type}", actor_id=npc.id, object_id=node.id, ) return f"gathering {node.resource_type}" return "searching for resources" if action in ("move_to_resource", "find_food", "find_herbs"): resource = _resource_from_directive(world, directive, npc) dest = destination or (resource.position if resource is not None else None) or _resource_destination(world, npc, action) if dest is not None: border = _border_block_summary(dest, half_width, half_depth) npc.position = move_toward( npc.position, dest, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth ) if border is not None: return border return "heading to resources" return "wandering" if action == "move_to": # Free wander in a deterministic-but-varied direction (anti-clumping). rng = random.Random(f"{world.seed}:{world.tick}:{npc.id}:wander") angle = rng.uniform(0.0, 2.0 * math.pi) dest = Vec3( x=npc.position.x + math.cos(angle) * NPC_MOVE_STEP * 3, y=0.0, z=npc.position.z + math.sin(angle) * NPC_MOVE_STEP * 3, ) npc.position = move_toward( npc.position, dest, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth ) return "wandering" if action == "go_home": house = nearest_house(world, npc.position) if house is not None: npc.home_house_id = house.id npc.position = move_toward( npc.position, house.position, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return f"going home to {house.id}" return "looking for home" if action == "patrol": target = _patrol_target(world, npc) if target is not None: npc.position = move_toward( npc.position, target, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return "patrolling" return "standing guard" if action == "build": return _apply_build(npc, world) if action == "craft": return _apply_craft(npc, world, directive) if action == "assign_cannon_operator": return _apply_assign_cannon_operator(npc, world, directive) if action == "fire": return _apply_fire_cannon(npc, world, directive) if action == "flee": beast = _beast_from_directive(world, npc, directive) if beast is not None: npc.position = move_away( npc.position, beast.position, FLEE_STEP, half_width=half_width, half_depth=half_depth, ) return "fleeing the beast" return "looking around" if action == "attack": target_npc = _melee_target_npc(world, npc, directive) if target_npc is not None: return _apply_npc_attack(npc, world, target_npc) beast = _beast_from_directive(world, npc, directive) if beast is None: return "defending" npc.focus_target_id = beast.id if vec_distance(npc.position, beast.position) <= NPC_MELEE_REACH: damage = _npc_damage_against_beast(npc) beast.health -= damage npc.relationships[beast.id] = -1.0 add_episode( npc, world.tick, "attack", npc.id, f"I struck {beast.id} for {damage:g} damage", target_id=beast.id, subject_kind="beast", perspective="self", tags=["danger", "beast"], weight=0.8, ) remember(npc, world.tick, f"You struck {beast.id} for {damage:g} damage.") _log_event( world, "npc_attack", f"{npc.name} attacked {beast.id} for {damage:g}", severity="warning", actor_id=npc.id, target_id=beast.id, ) if npc.help_target_id: helped = _npc_by_id(world, npc.help_target_id) if helped is not None and is_alive(helped): add_episode( helped, world.tick, "attack", npc.id, f"{npc.name} came to help against {beast.id}", target_id=beast.id, subject_kind="npc", perspective="witness", tags=["help", "danger", "beast"], weight=0.8, ) if beast.health <= 0: beast.state = "dead" npc.beasts_killed += 1 world.beasts_killed += 1 world.overseer_score += 2 _earn_coins(npc, world, COINS_PER_BEAST_KILL, f"slaying {beast.id}") _record_world_event( world, "beast_killed", f"{npc.name} killed {beast.id}", beast.position, radius=WITNESS_RADIUS, duration_ticks=5, ) _log_event( world, "beast_killed", f"{npc.name} killed {beast.id}", severity="good", actor_id=npc.id, target_id=beast.id, ) remember(npc, world.tick, f"You killed {beast.id}!") return "killed the beast" return "fighting the beast" npc.position = move_toward( npc.position, beast.position, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return "charging the beast" if action == "transfer": treasury = _treasury_from_directive(world, directive) if treasury is not None: return _apply_treasury_transfer(npc, world, treasury, directive) # Share food from whoever has a surplus with the neediest nearby neighbour, # so food flows from haves to have-nots instead of people starving alone. partner = _transfer_partner(world, npc, directive) resource_type = _resource_type_from_directive(directive, default="food") amount = _amount_from_directive(directive, default=1) can_transfer_food = ( resource_type == "food" and npc.inventory_food >= amount and partner is not None and (npc.inventory_food >= SHARE_FOOD_SURPLUS or partner.hunger > npc.hunger) and partner.inventory_food < npc.inventory_food ) can_transfer_nonfood = ( resource_type != "food" and partner is not None and _inventory_amount(npc, resource_type) >= amount ) if partner is not None and (can_transfer_food or can_transfer_nonfood): npc.focus_target_id = partner.id _remove_from_inventory(npc, resource_type, amount) _add_to_inventory_amount(partner, resource_type, amount) npc.resources_transferred += amount _adjust_relationship(npc, partner.id, 0.2) add_episode( npc, world.tick, "transfer", npc.id, f"I gave {amount} {resource_type} to {partner.name}", target_id=partner.id, object_id=resource_type, subject_kind="npc", perspective="self", tags=[resource_type, "help", "trade"], weight=0.5, ) add_episode( partner, world.tick, "transfer", npc.id, f"{npc.name} gave me {amount} {resource_type}", target_id=partner.id, object_id=resource_type, subject_kind="npc", perspective="recipient", tags=[resource_type, "help", "trade"], weight=0.7, ) remember(npc, world.tick, f"Gave {resource_type} to {partner.name}.") remember(partner, world.tick, f"Received {resource_type} from {npc.name}.") _log_event( world, "transfer", f"{npc.name} gave {amount} {resource_type} to {partner.name}", severity="good", actor_id=npc.id, target_id=partner.id, object_id=resource_type, ) return f"sharing {resource_type} with {partner.name}" return "looking for someone to trade with" if action == "steal": # Whom to rob is the model's call. Honor an explicit target (even a # countryman, if the model truly decides so); only fall back to the # nearest food-holder when no target was named. partner = _target_npc_from_directive(world, npc, directive, max_dist=TRADE_RADIUS) if partner is None: partner = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS, require_food=True) if partner is not None and partner.inventory_food > 0: npc.focus_target_id = partner.id partner.inventory_food -= 1 npc.inventory_food += 1 add_episode( partner, world.tick, "steal", npc.id, f"{npc.name} stole from me", target_id=partner.id, object_id="food", subject_kind="npc", perspective="recipient", tags=["food", "betrayal"], weight=0.9, ) add_episode( npc, world.tick, "steal", npc.id, f"I stole food from {partner.name}", target_id=partner.id, object_id="food", subject_kind="npc", perspective="self", tags=["food", "betrayal"], weight=0.7, ) remember(npc, world.tick, f"Stole food from {partner.name}.") remember(partner, world.tick, f"{npc.name} stole food from you.") return f"stealing from {partner.name}" return "looking to steal" if action == "communicate": return _apply_communicate(npc, world, directive) if action == "defend": return "defending" if action == "vote": return _apply_vote(npc, world, directive) return "idle" def _apply_vote(npc: Npc, world: WorldState, directive: NpcDirective | None) -> str: election = active_election_for_npc(world, npc) candidate_id = directive.target_npc_id if directive is not None else None if election is None or candidate_id not in election.candidate_ids: return "unable to vote" election.votes[npc.id] = candidate_id candidate = _npc_by_id(world, candidate_id) candidate_name = candidate.name if candidate is not None else candidate_id _log_event( world, "vote_cast", f"{npc.name} voted for {candidate_name}", actor_id=npc.id, target_id=candidate_id, object_id=election.country_id, ) return f"voting for {candidate_name}" def _apply_communicate( npc: Npc, world: WorldState, directive: NpcDirective | None, ) -> str: intent = _communication_intent(directive, npc) target = _target_npc_from_directive(world, npc, directive, max_dist=TALK_RADIUS) message = directive.message if directive and directive.message else _message_for_intent(intent) has_message = bool(directive is not None and directive.message) if intent == "help_request": add_episode( npc, world.tick, "communicate", npc.id, f"{npc.name} called for help", subject_kind="npc", perspective="self", tags=["help", "danger"], weight=0.6, ) helpers = resolve_call_for_help(npc, world) helper_ids = {helper.id for helper in helpers} for ally in npcs_in_radius(world, npc.position, HELP_RADIUS, exclude_id=npc.id): ally.fear = min(100.0, ally.fear + 5.0) if ally.id in helper_ids: continue add_episode( ally, world.tick, "communicate", npc.id, f"{npc.name} called for help", subject_kind="npc", perspective="recipient", tags=["help", "danger"], weight=0.6, ) _record_world_event( world, "communicate", f"{npc.name} called for help", npc.position, radius=HELP_RADIUS, duration_ticks=4, ) if helpers: names = ", ".join(helper.name for helper in helpers) return f"calling for help; {names} responded" return "calling for help" if intent == "trade_request": if target is None: target = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS, require_food=True) if target is None: return "looking for someone to trade with" npc.focus_target_id = target.id add_episode( npc, world.tick, "communicate", npc.id, f"I asked {target.name} for food", target_id=target.id, object_id="food", subject_kind="npc", perspective="self", tags=["food", "trade"], weight=0.4, ) add_episode( target, world.tick, "communicate", npc.id, f"{npc.name} asked me for food", target_id=target.id, object_id="food", subject_kind="npc", perspective="recipient", tags=["food", "trade"], weight=0.4, ) remember(npc, world.tick, f"You asked {target.name} for food.") remember(target, world.tick, f"{npc.name} asked you for food.") trust = target.relationships.get(npc.id, 0.0) if target.inventory_food > SHARE_FOOD_SURPLUS and trust > 0.3: transfer = NpcDirective( npc_id=target.id, action="transfer", target_npc_id=npc.id, resource_type="food", amount=1, ) return apply_action_effects(target, "transfer", world, directive=transfer) return f"requesting food from {target.name}: declined" if target is not None: npc.focus_target_id = target.id # Conversations read better when the listener turns toward the speaker too. target.focus_target_id = npc.id add_episode( npc, world.tick, "communicate", npc.id, f"I told {target.name}: {message}", target_id=target.id, subject_kind="npc", perspective="self", tags=["social"] if intent == "social" else [intent], weight=0.3, ) add_episode( target, world.tick, "communicate", npc.id, f"{npc.name} told me: {message}", target_id=target.id, subject_kind="npc", perspective="recipient", tags=["social"] if intent == "social" else [intent], weight=0.3, ) remember(npc, world.tick, f"You talked with {target.name}: {message}") remember(target, world.tick, f"{npc.name} talked with you: {message}") if has_message: # Bystanders within earshot overhear directed speech too. _overhear_speech(world, npc, message, intent, exclude_ids={target.id}) _log_event( world, "speech", f"{npc.name} to {target.name}: {message}", actor_id=npc.id, target_id=target.id, ) return f"talking to {target.name}" # No explicit listener: a shout heard by everyone within earshot. add_episode( npc, world.tick, "communicate", npc.id, f"I said: {message}", subject_kind="npc", perspective="self", tags=["social"] if intent == "social" else [intent], weight=0.3, ) remember(npc, world.tick, f"You said: {message}") heard_by = _overhear_speech(world, npc, message, intent, exclude_ids=set()) if has_message: _log_event(world, "speech", f"{npc.name}: {message}", actor_id=npc.id) if heard_by: return f"shouting (heard by {heard_by} nearby)" return "shouting, but no one is close enough to hear" def _emit_side_speech(world: WorldState, npc: Npc, message: str) -> int: """Emit speech said ALONGSIDE another action (the side-channel `speak` field). Resolved as a shout heard by everyone within earshot, mirroring how the dedicated speak action logs and remembers a line. Returns hearers count. """ text = message.strip() if not text: return 0 add_episode( npc, world.tick, "communicate", npc.id, f"I said: {text}", subject_kind="npc", perspective="self", tags=["social"], weight=0.3, ) remember(npc, world.tick, f"You said: {text}") heard = _overhear_speech(world, npc, text, "social", exclude_ids=set()) _log_event(world, "speech", f"{npc.name}: {text}", actor_id=npc.id) return heard def _overhear_speech( world: WorldState, speaker: Npc, message: str, intent: str, *, exclude_ids: set[str], ) -> int: """Record `message` in the memory/log of every NPC within earshot. Returns the number of NPCs who heard it. """ heard = 0 for hearer in npcs_in_radius(world, speaker.position, SPEECH_RADIUS, exclude_id=speaker.id): if hearer.id in exclude_ids: continue add_episode( hearer, world.tick, "communicate", speaker.id, f"{speaker.name} said: {message}", subject_kind="npc", perspective="witness", tags=["social"] if intent == "social" else [intent], weight=0.2, ) remember(hearer, world.tick, f"{speaker.name} said: {message}") heard += 1 return heard def _apply_build(npc: Npc, world: WorldState) -> str: half_width = world.terrain.width / 2 half_depth = world.terrain.depth / 2 # Priority 0: patch up a standing-but-damaged home before it collapses. # Defending the base outranks finishing a new build or hauling more wood. damaged = _nearest_damaged_house(world, npc.position) if damaged is not None and npc.inventory_wood >= HOUSE_REPAIR_WOOD_COST: if vec_distance(npc.position, damaged.position) > HOUSE_BUILD_RADIUS: npc.position = move_toward( npc.position, damaged.position, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return f"rushing to repair {damaged.id}" npc.inventory_wood -= HOUSE_REPAIR_WOOD_COST before = damaged.hp damaged.hp = min(damaged.max_hp, damaged.hp + HOUSE_REPAIR_PER_ACTION) _earn_coins(npc, world, COINS_PER_BUILD_TICK, f"repairing {damaged.id}") _log_event( world, "house_repaired", f"{npc.name} repaired {damaged.id} (+{round(damaged.hp - before)} hp)", severity="good", actor_id=npc.id, target_id=damaged.id, ) remember(npc, world.tick, f"You repaired {damaged.id}.") return f"repairing {damaged.id}" active_house = _house_by_id(world, npc.build_target_house_id or "") if active_house is not None and active_house.state == "under_construction": distance = vec_distance(npc.position, active_house.position) if distance > HOUSE_BUILD_RADIUS: npc.position = move_toward( npc.position, active_house.position, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return f"returning to build {active_house.id}" active_house.build_progress += 1 _earn_coins(npc, world, COINS_PER_BUILD_TICK, f"building {active_house.id}") if npc.id not in active_house.owner_ids: active_house.owner_ids.append(npc.id) if active_house.build_progress >= 10: active_house.state = "completed" active_house.hp = active_house.max_hp npc.build_target_house_id = None npc.houses_built += 1 world.houses_built += 1 world.overseer_score += 2 _earn_coins(npc, world, COINS_ON_BUILD_COMPLETE, f"completing {active_house.id}") _record_world_event( world, "build_completed", f"{npc.name} completed {active_house.id}", active_house.position, radius=WITNESS_RADIUS, duration_ticks=6, ) _log_event( world, "build_completed", f"{npc.name} completed {active_house.id}", severity="good", actor_id=npc.id, target_id=active_house.id, ) remember(npc, world.tick, f"You completed {active_house.id}.") return f"completed {active_house.id}" return f"building {active_house.id}" if npc.inventory_wood < 5: return "needs wood to build" ruin = _nearest_destroyed_house(world, npc.position) if ruin is not None: if vec_distance(npc.position, ruin.position) > HOUSE_BUILD_RADIUS: npc.position = move_toward( npc.position, ruin.position, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return f"heading to rebuild {ruin.id}" npc.inventory_wood -= 5 ruin.state = "under_construction" ruin.build_progress = 1 if npc.id not in ruin.owner_ids: ruin.owner_ids.append(npc.id) npc.build_target_house_id = ruin.id _log_event( world, "build_started", f"{npc.name} started rebuilding {ruin.id}", severity="good", actor_id=npc.id, target_id=ruin.id, ) remember(npc, world.tick, f"You started rebuilding {ruin.id}.") return f"rebuilding {ruin.id}" if not _can_start_house(world, npc.position): site = _preferred_build_site(world, npc) npc.position = move_toward( npc.position, site, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return "moving to build site" npc.inventory_wood -= 5 house = House( id=f"house_{_next_house_index(world):03d}", position=npc.position, owner_ids=[npc.id], hp=60.0, max_hp=60.0, state="under_construction", build_progress=1, capacity=3, ) world.houses.append(house) npc.build_target_house_id = house.id _record_world_event( world, "build_started", f"{npc.name} started {house.id}", house.position, radius=WITNESS_RADIUS, duration_ticks=5, ) _log_event( world, "build_started", f"{npc.name} started {house.id}", severity="good", actor_id=npc.id, target_id=house.id, ) remember(npc, world.tick, f"You started building {house.id}.") return f"building {house.id}" def _can_craft_cannon(world: WorldState, npc: Npc) -> bool: country = country_for_npc(world, npc) if country is None or country.ruler_id != npc.id or country.cannon is not None: return False if len(country.citizen_ids) < _env_int("CANNON_MIN_POPULATION", DEFAULT_CANNON_MIN_POPULATION): return False treasury = country.treasury.resources if country.treasury else {} return ( treasury.get("coins", 0) >= _env_int("CANNON_TREASURY_COST", DEFAULT_CANNON_TREASURY_COST) and treasury.get("wood", 0) >= _env_int("CANNON_WOOD_COST", DEFAULT_CANNON_WOOD_COST) ) def _can_assign_cannon_operator( world: WorldState, npc: Npc, directive: NpcDirective | None, ) -> bool: country = country_for_npc(world, npc) target_id = _param_str(directive, "npc_id") or (directive.target_npc_id if directive else None) target = _npc_by_id(world, target_id or "") return ( country is not None and country.ruler_id == npc.id and country.cannon is not None and target is not None and target.country_id == country.id ) def _can_fire_cannon(world: WorldState, npc: Npc, directive: NpcDirective | None) -> bool: country = country_for_npc(world, npc) if country is None or country.cannon is None: return False return country.cannon.operator_id == npc.id and world.tick >= country.cannon.cooldown_until_tick def _apply_craft(npc: Npc, world: WorldState, directive: NpcDirective | None) -> str: if _param_str(directive, "recipe_id") not in {None, "cannon"}: return "unknown recipe" if not _can_craft_cannon(world, npc): return "cannot craft cannon yet" country = country_for_npc(world, npc) assert country is not None and country.treasury is not None treasury = country.treasury.resources coin_cost = _env_int("CANNON_TREASURY_COST", DEFAULT_CANNON_TREASURY_COST) wood_cost = _env_int("CANNON_WOOD_COST", DEFAULT_CANNON_WOOD_COST) hp_tax = _env_int("CANNON_HP_TAX_PER_CITIZEN", DEFAULT_CANNON_HP_TAX_PER_CITIZEN) treasury["coins"] = treasury.get("coins", 0) - coin_cost treasury["wood"] = treasury.get("wood", 0) - wood_cost for citizen in world.living_npcs(): if citizen.country_id == country.id: citizen.health = max(1, citizen.health - hp_tax) country.cannon = CannonState( id=f"cannon_{country.id}", position=country.treasury.position, damage=_env_int("CANNON_DAMAGE", DEFAULT_CANNON_DAMAGE), radius=_env_int("CANNON_RADIUS", DEFAULT_CANNON_RADIUS), ) _log_event( world, "cannon_crafted", f"{country.name} crafted {country.cannon.id}", severity="good", actor_id=npc.id, object_id=country.cannon.id, ) return f"crafting {country.cannon.id}" def _apply_assign_cannon_operator( npc: Npc, world: WorldState, directive: NpcDirective | None, ) -> str: if not _can_assign_cannon_operator(world, npc, directive): return "cannot assign cannon operator" country = country_for_npc(world, npc) assert country is not None and country.cannon is not None target_id = _param_str(directive, "npc_id") or directive.target_npc_id country.cannon.operator_id = target_id _refresh_country_citizens(world, world.tick) _log_event( world, "cannon_operator_assigned", f"{country.name} assigned {target_id} to {country.cannon.id}", severity="good", actor_id=npc.id, target_id=target_id, object_id=country.cannon.id, ) return f"assigning cannon operator {target_id}" def _apply_fire_cannon(npc: Npc, world: WorldState, directive: NpcDirective | None) -> str: if not _can_fire_cannon(world, npc, directive): return "cannot fire cannon" country = country_for_npc(world, npc) assert country is not None and country.cannon is not None x = _param_float(directive, "x") z = _param_float(directive, "z") if x is None or z is None: return "missing cannon target" target = Vec3(x=x, y=0.0, z=z) damaged: list[str] = [] for target_npc in list(world.living_npcs()): if vec_distance(target_npc.position, target) > country.cannon.radius: continue target_npc.health -= country.cannon.damage damaged.append(target_npc.id) if target_npc.health <= 0: _kill_npc(world, target_npc, cause="cannon", actor_id=npc.id) cooldown = _env_int("CANNON_COOLDOWN_TICKS", DEFAULT_CANNON_COOLDOWN_TICKS) country.cannon.cooldown_until_tick = world.tick + cooldown _log_event( world, "cannon_fired", f"{npc.name} fired {country.cannon.id} at ({x:g}, {z:g}) hitting {len(damaged)} targets", severity="danger" if damaged else "warning", actor_id=npc.id, object_id=country.cannon.id, ) return f"firing cannon at {x:g}, {z:g}" def _param_str(directive: NpcDirective | None, key: str) -> str | None: if directive is None or not isinstance(directive.params, dict): return None value = directive.params.get(key) return value if isinstance(value, str) else None def _param_float(directive: NpcDirective | None, key: str) -> float | None: if directive is None or not isinstance(directive.params, dict): return None value = directive.params.get(key) if isinstance(value, int | float) and not isinstance(value, bool): return float(value) return None def _patrol_target(world: WorldState, npc: Npc) -> Vec3 | None: threat = _nearest_village_threat(world, npc) if threat is not None: return threat.position houses = completed_houses(world) if not houses: return Vec3(x=0.0, y=0.0, z=0.0) houses = sorted(houses, key=lambda house: house.id) if vec_distance(npc.position, houses[npc.patrol_index % len(houses)].position) <= NPC_MOVE_STEP: npc.patrol_index = (npc.patrol_index + 1) % len(houses) return houses[npc.patrol_index % len(houses)].position def _nearest_village_threat(world: WorldState, npc: Npc) -> Beast | None: candidates = [ beast for beast in world.beasts if beast.state != "dead" and ( any(vec_distance(beast.position, house.position) <= VILLAGE_THREAT_RADIUS for house in world.houses) or any( vec_distance(beast.position, other.position) <= VILLAGE_THREAT_RADIUS for other in world.living_npcs() ) ) ] if not candidates: return nearest_beast(world, npc.position) return min(candidates, key=lambda beast: (vec_distance(npc.position, beast.position), beast.id)) def _npc_damage_against_beast(npc: Npc) -> float: damage = BASE_NPC_DAMAGE + (WEAPON_DAMAGE_BONUS if npc.inventory_weapon > 0 else 0) role = normalize_role(npc.role) if role == "guard": return damage * GUARD_DAMAGE_MULTIPLIER if role in {"gatherer", "builder"}: return damage * WEAK_DAMAGE_MULTIPLIER return float(damage) def _melee_target_npc( world: WorldState, npc: Npc, directive: NpcDirective | None, ) -> Npc | None: """The living NPC this attack directive explicitly targets (any nation). Whether to strike — and whom, even a countryman — is the model's call, not the engine's. We only require an explicit, living target other than self; we never auto-pick someone to attack, so allies are not hit by accident. """ if directive is None: return None target_id = directive.target_npc_id or directive.target_entity_id if not target_id: return None target = _npc_by_id(world, target_id) if target is None or target.id == npc.id or not is_alive(target): return None return target def _apply_npc_attack(npc: Npc, world: WorldState, target: Npc) -> str: """Melee strike against a rival NPC: close in if out of reach, else wound them.""" half_width = world.terrain.width / 2 half_depth = world.terrain.depth / 2 npc.focus_target_id = target.id if vec_distance(npc.position, target.position) > NPC_MELEE_REACH: npc.position = move_toward( npc.position, target.position, NPC_MOVE_STEP, half_width=half_width, half_depth=half_depth, ) return f"charging {target.name}" damage = _npc_damage_against_beast(npc) target.health -= damage target.fear = min(100.0, max(target.fear, FEAR_ON_ATTACK)) npc.relationships[target.id] = -1.0 add_episode( npc, world.tick, "attack", npc.id, f"I attacked {target.name} for {damage:g} damage", target_id=target.id, subject_kind="npc", perspective="self", tags=["combat", "raid"], weight=0.9, ) add_episode( target, world.tick, "attack", npc.id, f"{npc.name} attacked me", target_id=target.id, subject_kind="npc", perspective="recipient", tags=["danger", "combat", "betrayal"], weight=1.0, ) remember(npc, world.tick, f"You attacked {target.name}.") remember(target, world.tick, f"{npc.name} attacked you at tick {world.tick}.") _record_world_event( world, "npc_attack", f"{npc.name} attacked {target.name}", target.position, radius=WITNESS_RADIUS, duration_ticks=4, ) _log_event( world, "npc_attack", f"{npc.name} attacked {target.name} for {damage:g}", severity="danger", actor_id=npc.id, target_id=target.id, ) for witness in npcs_in_radius(world, target.position, WITNESS_RADIUS, exclude_id=target.id): if witness.id == npc.id: continue witness.fear = min(100.0, witness.fear + FEAR_ON_WITNESS) add_episode( witness, world.tick, "attack", npc.id, f"{npc.name} attacked {target.name} nearby", target_id=target.id, subject_kind="npc", perspective="witness", tags=["danger", "combat"], weight=0.8, ) if target.health <= 0: _kill_npc(world, target, cause="slain", actor_id=npc.id) return f"killed {target.name}" return f"attacking {target.name}" def _house_by_id(world: WorldState, house_id: str) -> House | None: if not house_id: return None return next((house for house in world.houses if house.id == house_id), None) def _nearest_destroyed_house(world: WorldState, position: Vec3) -> House | None: ruins = [house for house in world.houses if house.state == "destroyed"] if not ruins: return None return min(ruins, key=lambda house: (vec_distance(position, house.position), house.id)) def _nearest_damaged_house(world: WorldState, position: Vec3) -> House | None: """Nearest standing home that has taken damage but is not yet destroyed.""" damaged = [ house for house in world.houses if house.state == "completed" and 0 < house.hp < house.max_hp ] if not damaged: return None return min(damaged, key=lambda house: (vec_distance(position, house.position), house.id)) def _can_start_house(world: WorldState, position: Vec3) -> bool: return all(vec_distance(position, house.position) >= HOUSE_MIN_DISTANCE for house in world.houses) def _preferred_build_site(world: WorldState, npc: Npc) -> Vec3: half_width = world.terrain.width / 2 half_depth = world.terrain.depth / 2 angle = (len(world.houses) * 1.61803398875) % (2 * math.pi) center = nearest_house(world, npc.position, completed_only=False) origin = center.position if center is not None else Vec3(x=0.0, y=0.0, z=0.0) return Vec3( x=round(_clamp(origin.x + math.cos(angle) * (HOUSE_MIN_DISTANCE + TILE), -half_width, half_width), 3), y=0.0, z=round(_clamp(origin.z + math.sin(angle) * (HOUSE_MIN_DISTANCE + TILE), -half_depth, half_depth), 3), ) def _next_house_index(world: WorldState) -> int: highest = 0 for house in world.houses: suffix = house.id.removeprefix("house_") if suffix.isdigit(): highest = max(highest, int(suffix)) return highest + 1 def resolve_call_for_help(caller: Npc, world: WorldState) -> list[Npc]: helpers: list[Npc] = [] for helper in npcs_in_radius(world, caller.position, HELP_RADIUS, exclude_id=caller.id): if helper.inventory_weapon <= 0: continue if helper.relationships.get(caller.id, 0.0) <= 0.3: continue helper.survival_goal = "help_ally" helper.help_target_id = caller.id add_episode( helper, world.tick, "communicate", caller.id, f"{caller.name} called for help", subject_kind="npc", perspective="recipient", tags=["help", "danger"], weight=0.6, ) helpers.append(helper) return helpers def _resource_from_directive( world: WorldState, directive: NpcDirective | None, npc: Npc, *, max_dist: float | None = None, ) -> ResourceNode | None: if directive is not None and directive.resource_id: node = next((node for node in world.resource_nodes if node.id == directive.resource_id), None) if node is not None and (max_dist is None or vec_distance(npc.position, node.position) <= max_dist): return node return nearest_resource_for_npc(world, npc, max_dist=max_dist) def _beast_from_directive( world: WorldState, npc: Npc, directive: NpcDirective | None, ) -> Beast | None: if directive is not None and directive.target_entity_id: beast = next( ( beast for beast in world.beasts if beast.id == directive.target_entity_id and beast.state != "dead" ), None, ) if beast is not None: return beast return nearest_beast(world, npc.position) def _target_npc_from_directive( world: WorldState, npc: Npc, directive: NpcDirective | None, *, max_dist: float | None = None, ) -> Npc | None: if directive is None or directive.target_npc_id is None: return None target = _npc_by_id(world, directive.target_npc_id) if target is None or target.id == npc.id or not is_alive(target): return None if max_dist is not None and vec_distance(npc.position, target.position) > max_dist: return None return target def _transfer_partner(world: WorldState, npc: Npc, directive: NpcDirective | None) -> Npc | None: partner = _target_npc_from_directive(world, npc, directive, max_dist=TRADE_RADIUS) if partner is not None: return partner return _neediest_food_neighbor(world, npc) def _treasury_from_directive( world: WorldState, directive: NpcDirective | None, ) -> tuple[CountryState, TreasuryState] | None: target_id = directive.target_entity_id if directive is not None else None if target_id is None: return None for country in world.countries: if country.treasury is not None and country.treasury.id == target_id: return country, country.treasury return None def _apply_treasury_transfer( npc: Npc, world: WorldState, treasury_ref: tuple[CountryState, TreasuryState], directive: NpcDirective | None, ) -> str: country, treasury = treasury_ref resource_type = _resource_type_from_directive(directive, default="coins") amount = _amount_from_directive(directive, default=1) take = bool(directive and directive.take) position = treasury.position resources = treasury.resources if vec_distance(npc.position, position) > TRADE_RADIUS: npc.position = move_toward( npc.position, position, NPC_MOVE_STEP, half_width=world.terrain.width / 2, half_depth=world.terrain.depth / 2, ) return "approaching treasury" if take: if npc.country_id == country.id: return "guarding own treasury" available = resources.get(resource_type, 0) if available <= 0: return "found empty treasury" moved = min(amount, available) resources[resource_type] = available - moved _add_to_inventory_amount(npc, resource_type, moved) _log_event( world, "treasury_stolen", f"{npc.name} stole {moved} {resource_type} from {country.name}", severity="warning", actor_id=npc.id, object_id=treasury.id, ) # Every living member of the robbed country learns who took from their # treasury, so they can confront the thief (grievance -> demand -> fight). for member in world.living_npcs(): if member.country_id != country.id or member.id == npc.id: continue add_episode( member, world.tick, "steal", npc.id, f"{npc.name} stole {moved} {resource_type} from our treasury", target_id=treasury.id, object_id=resource_type, subject_kind="npc", perspective="recipient", tags=[resource_type, "betrayal", "treasury"], weight=0.9, ) remember( member, world.tick, f"{npc.name} stole {moved} {resource_type} from our treasury.", ) return f"stealing {resource_type} from {country.name}" if npc.country_id != country.id: return "refusing to fund foreign treasury" available = _inventory_amount(npc, resource_type) if available <= 0: return f"no {resource_type} to deposit" moved = min(amount, available) _remove_from_inventory(npc, resource_type, moved) resources[resource_type] = resources.get(resource_type, 0) + moved npc.resources_transferred += moved _log_event( world, "treasury_deposit", f"{npc.name} deposited {moved} {resource_type} into {country.name} treasury", severity="good", actor_id=npc.id, object_id=treasury.id, ) return f"depositing {resource_type} into {country.name} treasury" def _resource_type_from_directive(directive: NpcDirective | None, *, default: str) -> str: if directive is not None and directive.resource_type in {"food", "herbs", "wood", "weapon", "coins"}: return directive.resource_type return default def _amount_from_directive(directive: NpcDirective | None, *, default: int) -> int: if directive is not None and directive.amount is not None: return max(1, min(20, directive.amount)) return default def _communication_intent(directive: NpcDirective | None, npc: Npc) -> str: raw = None if directive is not None: raw = directive.communication_intent or directive.intent if isinstance(raw, str): if "help_request" in raw or "call_for_help" in raw: return "help_request" if "survive_threat" in raw: return "help_request" if "trade_request" in raw or "request_trade" in raw: return "trade_request" if "warning" in raw: return "warning" if "social" in raw or "talk" in raw: return "social" if npc.fear > 45: return "help_request" if npc.hunger > STARVATION_THRESHOLD: return "trade_request" return "social" def _message_for_intent(intent: str) -> str: if intent == "help_request": return "Help! I am in danger!" if intent == "trade_request": return "Can you spare food?" if intent == "warning": return "Stay alert." return "How are you holding up?" def _earn_coins(npc: Npc, world: WorldState, amount: int, reason: str) -> None: """Pay an NPC a coin wage for productive work and log it as income.""" if amount <= 0: return npc.inventory_coins += amount _log_event( world, "coins_earned", f"{npc.name} earned {amount} coins by {reason}", severity="good", actor_id=npc.id, ) def _add_to_inventory(npc: Npc, resource_type: str) -> None: _add_to_inventory_amount(npc, resource_type, 1) def _add_to_inventory_amount(npc: Npc, resource_type: str, amount: int) -> None: if resource_type == "food": npc.inventory_food += amount elif resource_type == "herbs": npc.inventory_herbs += amount elif resource_type == "wood": npc.inventory_wood += amount elif resource_type == "weapon": npc.inventory_weapon += amount elif resource_type == "coins": npc.inventory_coins += amount def _remove_from_inventory(npc: Npc, resource_type: str, amount: int) -> None: if resource_type == "food": npc.inventory_food = max(0, npc.inventory_food - amount) elif resource_type == "herbs": npc.inventory_herbs = max(0, npc.inventory_herbs - amount) elif resource_type == "wood": npc.inventory_wood = max(0, npc.inventory_wood - amount) elif resource_type == "weapon": npc.inventory_weapon = max(0, npc.inventory_weapon - amount) elif resource_type == "coins": npc.inventory_coins = max(0, npc.inventory_coins - amount) def _inventory_amount(npc: Npc, resource_type: str) -> int: if resource_type == "food": return npc.inventory_food if resource_type == "herbs": return npc.inventory_herbs if resource_type == "wood": return npc.inventory_wood if resource_type == "weapon": return npc.inventory_weapon if resource_type == "coins": return npc.inventory_coins return 0 def _resource_destination(world: WorldState, npc: Npc, action: str) -> Vec3 | None: if normalize_role(npc.role) == "builder": resource_type: str | None = "wood" elif action == "find_food": resource_type: str | None = "food" elif action == "find_herbs": resource_type = "herbs" else: resource_type = None node = nearest_resource_for_npc(world, npc, resource_type=resource_type) if node is None: node = nearest_resource_for_npc(world, npc) return node.position if node is not None else None def _spread_resource( world: WorldState, npc: Npc, index: int, *, resource_type: str | None = None, ) -> ResourceNode | None: """Pick one of the nearest non-empty nodes, fanned out by NPC index. Spreading foragers across several nearby nodes stops the whole settlement from piling onto a single node and then starving together. """ if normalize_role(npc.role) == "builder": resource_type = "wood" nodes = sorted( ( node for node in world.resource_nodes if node.amount > 0 and (resource_type is None or node.resource_type == resource_type) ), key=lambda node: (vec_distance(npc.position, node.position), node.id), ) if not nodes and resource_type is not None and normalize_role(npc.role) != "builder": nodes = sorted( (node for node in world.resource_nodes if node.amount > 0), key=lambda node: (vec_distance(npc.position, node.position), node.id), ) if not nodes: return None return nodes[index % min(SPREAD_NODE_CHOICES, len(nodes))] def _neediest_food_neighbor(world: WorldState, npc: Npc) -> Npc | None: candidates = [ other for other in world.living_npcs() if other.id != npc.id and vec_distance(npc.position, other.position) <= TRADE_RADIUS ] if not candidates: return None return min(candidates, key=lambda other: (other.inventory_food, -other.hunger, other.id)) def _small_talk_line(npc: Npc, partner: Npc) -> str: if npc.fear > 30: return "stay alert, the beast is near" if npc.hunger > 60: return "I'm getting hungry" if npc.inventory_food >= SHARE_FOOD_SURPLUS: return "I have food to spare if you need it" if npc.inventory_weapon > 0: return "keep close, I'm armed" return "how are you holding up?" def _npc_by_id(world: WorldState, npc_id: str) -> Npc | None: return next((candidate for candidate in world.living_npcs() if candidate.id == npc_id), None) def _adjust_relationship(npc: Npc, target_id: str, delta: float) -> None: current = -1.0 if target_id.startswith("beast") else npc.relationships.get(target_id, 0.0) npc.relationships[target_id] = round(_clamp(current + delta, -1.0, 1.0), 3) def _has_trusted_ally_nearby(world: WorldState, npc: Npc) -> bool: for other in npcs_in_radius(world, npc.position, ALLY_RADIUS, exclude_id=npc.id): trust = npc.relationships.get(other.id, 0.0) if trust > 0.3: return True return False def _relationship_label(trust: float) -> str: if trust >= 0.5: return "ally" if trust <= -0.3: return "enemy" return "neutral" def _memory_tag(episode: MemoryEpisode) -> str: if "danger" in episode.tags or episode.kind in {"attack", "steal"}: return "high threat" if episode.emotional_weight >= 0.8 else "threat" if "help" in episode.tags or episode.kind == "transfer": return "ally" if "food" in episode.tags: return "food" return "memory" def _direction(origin: Vec3, target: Vec3) -> str: # Convention matches the briefing compass: +X=WEST, -X=EAST, +Z=NORTH, -Z=SOUTH. dx = target.x - origin.x dz = target.z - origin.z if abs(dx) >= abs(dz): return "WEST" if dx >= 0 else "EAST" return "NORTH" if dz >= 0 else "SOUTH" def _tiles(distance: float) -> float: return distance / TILE def _clamp(value: float, minimum: float, maximum: float) -> float: return min(max(value, minimum), maximum) # --------------------------------------------------------------------------- # # Deterministic survival planner + resolver # --------------------------------------------------------------------------- # def is_survival_world(world: WorldState) -> bool: return bool(world.beasts or world.resource_nodes or world.houses) def survival_directive_for(world: WorldState, npc: Npc, next_tick: int) -> NpcDirective: """Deterministic survival directive for a single NPC (used as a safe fallback).""" goal = select_goal(npc, world) index = next((i for i, candidate in enumerate(world.npcs) if candidate.id == npc.id), 0) return _plan_directive(world, npc, goal, next_tick, index) def propose_survival_tick(world: WorldState, next_tick: int) -> TickPlan: """Deterministic survival plan: one survival action per living NPC.""" directives: list[NpcDirective] = [] for index, npc in enumerate(world.living_npcs()): goal = select_goal(npc, world) directives.append(_plan_directive(world, npc, goal, next_tick, index)) return TickPlan(source="survival", directives=directives) def apply_survival_plan( world: WorldState, next_tick: int, plan: TickPlan, ) -> list[dict[str, object]]: """Resolve a survival plan onto the live world. Returns debug traces.""" debug: list[dict[str, object]] = [] npcs_by_id = {npc.id: npc for npc in world.npcs} for directive in plan.directives: npc = npcs_by_id.get(directive.npc_id) if npc is None: continue if not is_alive(npc): npc.intention = "dead" continue requested_effect = _canonical_action(_verb_for_action(npc, directive.action, world, directive)) action = validate_survival_action(npc, directive.action, world, directive) summary = apply_action_effects( npc, action, world, destination=directive.target, directive=directive, ) npc.intention = summary # Side-channel speech: words said ALONGSIDE a non-speak action. The speak # action itself already emits its message, so skip it to avoid doubling. if directive.speech and directive.action != "speak": _emit_side_speech(world, npc, directive.speech) # Surface the model's chain-of-thought on the live NPC for the UI. Keep # the last real thought when a turn produced none (e.g. forced retry). if directive.reasoning: npc.last_reasoning = directive.reasoning npc.last_reasoning_tick = next_tick if directive.conversation_user and directive.conversation_assistant: npc.conversation_context.add_turn( directive.conversation_user, directive.conversation_assistant, ) if directive.memory: remember(npc, next_tick, directive.memory) # The NPC owns its own long-term memory: when the model rewrote its memory # note this turn, that REPLACES the engine's mechanical tick dump. Set it # LAST, after every remember() above, so compaction can't clobber it. # Strip any "Tick N:" labels - the note is prose, never a tick log. if directive.memory_summary: cleaned = strip_tick_entries(directive.memory_summary) if cleaned: npc.memory_summary = cleaned npc.memory_summary_self_authored = True if next_tick > 0 and next_tick % 10 == 0: compress_to_facts(npc) debug.append( { "npc_id": npc.id, "goal": npc.survival_goal, "requested_action": directive.action, "requested_effect": requested_effect, "action": action, "summary": summary, "validator_verdict": _survival_validator_verdict( requested_action=str(directive.action), requested_effect=requested_effect, resolved_action=action, ), } ) for npc in world.npcs: if not is_alive(npc): npc.intention = "dead" return debug def _survival_validator_verdict( *, requested_action: str, requested_effect: str, resolved_action: str, ) -> dict[str, object]: if requested_effect == resolved_action: return { "status": "accepted", "requested": requested_action, "requested_effect": requested_effect, "resolved": resolved_action, "reason": "accepted", } if resolved_action == "idle" and requested_effect != "idle": return { "status": "rejected", "requested": requested_action, "requested_effect": requested_effect, "resolved": resolved_action, "reason": "invalid or impossible survival action", } return { "status": "repaired", "requested": requested_action, "requested_effect": requested_effect, "resolved": resolved_action, "reason": "engine validation repaired action", } def _plan_directive( world: WorldState, npc: Npc, goal: str, next_tick: int, index: int, ) -> NpcDirective: npc.survival_goal = goal def directive( action: str, *, target: Vec3 | None = None, target_npc_id: str | None = None, target_entity_id: str | None = None, resource_id: str | None = None, resource_type: str | None = None, amount: int | None = None, communication_intent: str | None = None, message: str | None = None, params: dict[str, object] | None = None, away: bool = False, ) -> NpcDirective: public_action = action public_use_type: str | None = None public_take = False if action in {"strike", "attack"}: public_action = "attack" elif action in {"gather", "consume", "eat", "heal", "build", "repair"}: public_action = "use" public_use_type = { "consume": "eat", "eat": "eat", "heal": "heal", "gather": "gather", "build": "build", "repair": "repair", }[action] if action in {"consume", "eat"} and resource_type is None: resource_type = "food" if action == "heal" and resource_type is None: resource_type = "herbs" elif action in {"craft", "assign_cannon_operator", "fire"}: public_action = "use" public_use_type = action elif action in {"communicate", "trade", "request_trade"}: public_action = "speak" elif action == "steal": public_action = "transfer" public_take = True elif action in { "go_home", "patrol", "flee", "move_to_resource", "move_to", "find_food", "find_herbs", "wander", }: public_action = "move" if action == "go_home" and target is None: home = nearest_house(world, npc.position) target = home.position if home is not None else None elif action == "patrol" and target is None: target = _patrol_target(world, npc) away = away or action == "flee" return NpcDirective( npc_id=npc.id, action=public_action, # type: ignore[arg-type] target=target, target_npc_id=target_npc_id, target_entity_id=target_entity_id, resource_id=resource_id, resource_type=resource_type, use_type=public_use_type, params=params, amount=amount, communication_intent=communication_intent, message=message, away=away, take=public_take, intent=f"survival:{goal}", ) role = normalize_role(npc.role) election = active_election_for_npc(world, npc, tick=next_tick) if election is not None and npc.id not in election.votes and election.candidate_ids: candidate_id = _fallback_vote(world, election, voter_id=npc.id) return directive("vote", target_npc_id=candidate_id) country = country_for_npc(world, npc) if country is not None and country.ruler_id == npc.id: if country.cannon is None and _can_craft_cannon(world, npc): return directive("craft", params={"recipe_id": "cannon"}) if country.cannon is not None and country.cannon.operator_id is None: operator = next( ( citizen for citizen in world.living_npcs() if citizen.country_id == country.id and citizen.id != npc.id ), npc, ) return directive("assign_cannon_operator", params={"npc_id": operator.id}) if ( country is not None and country.cannon is not None and country.cannon.operator_id == npc.id and next_tick >= country.cannon.cooldown_until_tick ): target = next( (other for other in world.living_npcs() if other.country_id != country.id), None, ) if target is not None: return directive( "fire", target=country.cannon.position, target_npc_id=target.id, params={"x": target.position.x, "z": target.position.z}, ) if goal == "obey_directive": text = (npc.god_directive or "").lower() target = _directive_named_target(world, npc, text) if any(word in text for word in ("attack", "kill", "fight", "hunt", "strike")): if target is not None: return directive("strike", target_npc_id=target.id) if "beast" in text or "monster" in text: beast = nearest_beast(world, npc.position) if beast is not None: return directive("strike", target=beast.position) return directive("strike") if any(word in text for word in ("home", "shelter", "hide", "safe")): return directive("go_home") if any(word in text for word in ("build", "repair", "house")): return directive("build") if any(word in text for word in ("wood", "food", "herb", "gather", "resource")): resource_type = "wood" if "wood" in text else "food" if "food" in text else "herbs" if "herb" in text else None node = _spread_resource(world, npc, index, resource_type=resource_type) if node is not None and vec_distance(npc.position, node.position) <= GATHER_RADIUS: return directive("use", resource_id=node.id, resource_type=resource_type) if node is not None: return directive("move", target=node.position, resource_id=node.id, resource_type=resource_type) if any(word in text for word in ("help", "warn", "talk", "speak", "tell")): partner = target or nearest_alive_npc(world, npc, max_dist=TALK_RADIUS) if partner is not None: return directive("speak", target_npc_id=partner.id, communication_intent="directive") if any(word in text for word in ("patrol", "guard", "watch")): return directive("patrol") return directive("idle") if goal == "engage_threat": beast = _nearest_village_threat(world, npc) if beast is not None: return directive("attack", target=beast.position, resource_id=None) return directive("patrol") if goal == "survive_threat_fight": if (next_tick + index) % 3 == 0: return directive("speak", communication_intent="help_request") return directive("strike") if goal == "survive_threat_flee": if (next_tick + index) % 4 == 0: return directive("speak", communication_intent="help_request") return directive("move", away=True) if goal == "help_ally": return directive("strike") if goal == "heal_self": return directive("use", resource_type="herbs") if goal == "eat_food": return directive("use", resource_type="food") if goal in ("find_food", "find_herbs"): resource_type = "food" if goal == "find_food" else "herbs" if role == "builder" and goal == "find_food": partner = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS, require_food=True) if partner is not None: return directive( "communicate", target_npc_id=partner.id, resource_type="food", communication_intent="trade_request", ) return directive("go_home") if goal == "find_food": partner = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS, require_food=True) if ( partner is not None and npc.inventory_food == 0 and partner.inventory_food > SHARE_FOOD_SURPLUS and npc.relationships.get(partner.id, 0.0) > 0.3 ): return directive( "speak", target_npc_id=partner.id, resource_type="food", communication_intent="trade_request", ) node = _spread_resource(world, npc, index, resource_type=resource_type) if node is not None: if vec_distance(npc.position, node.position) <= GATHER_RADIUS: return directive("use", resource_id=node.id) return directive("move", target=node.position, resource_id=node.id) return directive("move") # nothing to gather right now -> wander, don't freeze if goal == "settle": house = house_containing_npc(world, npc) if ( house is not None and _is_reproduction_candidate(npc) and REPRODUCTION_MAX_HUNGER < npc.hunger <= REPRODUCTION_SNACK_HUNGER and npc.inventory_food > 0 ): return directive("consume") if house is None: return directive("go_home") if (next_tick + index) % 3 == 0: neighbor = nearest_alive_npc(world, npc, max_dist=TALK_RADIUS) if neighbor is not None: return directive( "communicate", target_npc_id=neighbor.id, communication_intent="social", ) return directive("go_home") if goal == "work": if role == "guard": return directive("patrol") if role == "builder": damaged = _nearest_damaged_house(world, npc.position) if damaged is not None and npc.inventory_wood >= HOUSE_REPAIR_WOOD_COST: return directive("build") # _apply_build repairs the damaged home first if npc.build_target_house_id or npc.inventory_wood >= 5: return directive("build") node = _spread_resource(world, npc, index, resource_type="wood") if node is not None and vec_distance(npc.position, node.position) <= GATHER_RADIUS: return directive("gather", resource_id=node.id) if node is not None: return directive("move_to_resource", target=node.position, resource_id=node.id) return directive("go_home") builder = _nearest_role(world, npc, "builder", max_dist=TRADE_RADIUS) if builder is not None and npc.inventory_wood > 0: return directive( "transfer", target_npc_id=builder.id, resource_type="wood", amount=1, ) # routine_life/work fallback: socialise, share, forage, or wander -- and spread out. neighbor = nearest_alive_npc(world, npc, max_dist=TALK_RADIUS) phase = (next_tick + index) % 4 country = country_for_npc(world, npc) if country is not None and country.treasury is not None and phase == 2: if npc.inventory_coins > 0: return directive( "transfer", target=country.treasury.position, target_entity_id=country.treasury.id, resource_type="coins", amount=1, ) if npc.inventory_wood > 0: return directive( "transfer", target=country.treasury.position, target_entity_id=country.treasury.id, resource_type="wood", amount=1, ) if ( neighbor is not None and npc.inventory_food >= SHARE_FOOD_SURPLUS and neighbor.inventory_food < npc.inventory_food and phase == 0 ): return directive( "transfer", target_npc_id=neighbor.id, resource_type="food", amount=1, ) if neighbor is not None and phase == 1: return directive("speak", target_npc_id=neighbor.id, communication_intent="social") node = _spread_resource(world, npc, index) if node is not None and vec_distance(npc.position, node.position) <= GATHER_RADIUS: return directive("use", resource_id=node.id) if node is not None and phase != 3: return directive("move", target=node.position, resource_id=node.id) return directive("move") # wander to disperse def _directive_named_target(world: WorldState, npc: Npc, text: str) -> Npc | None: for other in world.living_npcs(): if other.id == npc.id: continue if other.id.lower() in text or other.name.lower() in text: return other return None