""" Generic construction behaviors for WorldSmithAI. This module implements domain-agnostic construction behaviors for arbitrary agent-based worlds. It deliberately avoids domain-specific classes such as Building, Road, Castle, Habitat, Factory, or PowerPlant. Instead, structures are represented as generic mutable records stored in an agent's state or memory. This keeps the simulation DSL-driven and allows an SLM to define many types of constructed entities without changing Python code. Implemented behaviors: - build: create or progress a generic structure record. - repair: restore durability on an existing structure. - expand: increase a generic structure dimension such as capacity or size. - demolish: damage, remove, or mark a structure as demolished. Example: behavior = BuildBehavior( structure_id="lab_1", structure_type="research_lab", material_cost={"materials": 2.0}, work_required=5.0, work_amount=1.0, initial_capacity=3.0, ) outcome = behavior.execute(agent, world) Future extensibility: - Add dependency graphs between structures. - Add construction crews and multi-agent work pooling. - Add terrain constraints. - Add world-level structure registries. - Add event emission for completed, damaged, expanded, or demolished assets. - Add graph infrastructure for roads, power grids, transport links, and networks. """ from __future__ import annotations import copy import logging from collections.abc import Iterable, Mapping, MutableMapping, MutableSequence, Sequence from dataclasses import dataclass, field from enum import Enum from numbers import Real from types import MappingProxyType from typing import TYPE_CHECKING, Any, ClassVar import numpy as np from core.behavior import Behavior if TYPE_CHECKING: from core.agent import Agent from core.world import World logger = logging.getLogger(__name__) Position = tuple[float, ...] class StorageLocation(str, Enum): """Supported mutable storage locations on an agent.""" STATE = "state" MEMORY = "memory" class StructureStatus(str, Enum): """Generic lifecycle statuses for constructed structures.""" PLANNED = "planned" ACTIVE = "active" COMPLETE = "complete" DAMAGED = "damaged" EXPANDED = "expanded" DEMOLISHED = "demolished" @dataclass(frozen=True) class ConstructionOutcome: """Serializable result returned by construction behavior execution. The core engine may ignore this result, but downstream components such as metrics, event streams, visualization, debugging tools, and narrators can consume it. """ behavior: str actor_id: str success: bool target_ids: tuple[str, ...] = () structure_ids: tuple[str, ...] = () step: int | None = None details: Mapping[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly representation of the outcome.""" return { "behavior": self.behavior, "actor_id": self.actor_id, "success": self.success, "target_ids": list(self.target_ids), "structure_ids": list(self.structure_ids), "step": self.step, "details": copy.deepcopy(dict(self.details)), } @dataclass(frozen=True) class StructureHandle: """Mutable reference to a structure record owned by an agent.""" owner: Agent structure_id: str record: MutableMapping[str, Any] store: MutableMapping[str, Any] location: StorageLocation store_key: str def _agent_id(agent: Agent) -> str: """Return a stable string identifier for an agent.""" return str(getattr(agent, "id")) def _agent_type(agent: Agent) -> str: """Return the generic type label for an agent.""" return str(getattr(agent, "type", "")) def _is_alive(agent: Agent) -> bool: """Return whether an agent can participate in behavior execution.""" return bool(getattr(agent, "alive", True)) def _world_step(world: World) -> int | None: """Return the current world step if available.""" value = getattr(world, "step_count", None) if isinstance(value, Real) and not isinstance(value, bool): return int(value) return None def _is_number(value: Any) -> bool: """Return whether a value is a real numeric scalar, excluding booleans.""" return isinstance(value, (Real, np.integer, np.floating)) and not isinstance(value, bool) def _as_float(value: Any, default: float = 0.0) -> float: """Safely convert a numeric-like value to float.""" if _is_number(value): return float(value) return default def _as_int(value: Any, default: int = 0) -> int: """Safely convert a numeric-like value to int.""" if _is_number(value): return int(value) return default def _normalize_storage(location: StorageLocation | str) -> StorageLocation: """Normalize a storage location value.""" if isinstance(location, StorageLocation): return location return StorageLocation(str(location)) def _agent_state(agent: Agent) -> MutableMapping[str, Any]: """Return an agent's mutable state mapping, creating one when needed.""" state = getattr(agent, "state", None) if isinstance(state, MutableMapping): return state replacement: dict[str, Any] = {} setattr(agent, "state", replacement) return replacement def _agent_memory(agent: Agent) -> MutableMapping[str, Any]: """Return an agent's mutable memory mapping, creating one when needed.""" memory = getattr(agent, "memory", None) if isinstance(memory, MutableMapping): return memory replacement: dict[str, Any] = {} setattr(agent, "memory", replacement) return replacement def _container_for(agent: Agent, location: StorageLocation | str) -> MutableMapping[str, Any]: """Return an agent container for a state-or-memory storage location.""" normalized = _normalize_storage(location) if normalized is StorageLocation.STATE: return _agent_state(agent) if normalized is StorageLocation.MEMORY: return _agent_memory(agent) raise ValueError(f"Unsupported storage location: {location!r}") def _ensure_mapping(parent: MutableMapping[str, Any], key: str) -> MutableMapping[str, Any]: """Return a nested mutable mapping under ``key``, creating one if absent.""" value = parent.get(key) if isinstance(value, MutableMapping): return value if isinstance(value, Mapping): replacement = dict(value) parent[key] = replacement return replacement replacement: dict[str, Any] = {} parent[key] = replacement return replacement def _ensure_list(parent: MutableMapping[str, Any], key: str) -> MutableSequence[Any]: """Return a nested mutable sequence under ``key``, creating one if absent.""" value = parent.get(key) if isinstance(value, MutableSequence): return value replacement: list[Any] = [] parent[key] = replacement return replacement def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None: """Append an item while enforcing an optional maximum history length.""" items.append(value) if max_items > 0 and len(items) > max_items: del items[: len(items) - max_items] def _split_path(path: str) -> tuple[str, ...]: """Split a dot-separated state or memory key into path components.""" return tuple(part for part in path.split(".") if part) def _get_path(container: Mapping[str, Any], path: str, default: Any = None) -> Any: """Read a possibly nested value from a mapping using dot notation.""" parts = _split_path(path) if not parts: return default current: Any = container for part in parts: if not isinstance(current, Mapping) or part not in current: return default current = current[part] return current def _set_path(container: MutableMapping[str, Any], path: str, value: Any) -> None: """Write a possibly nested value to a mapping using dot notation.""" parts = _split_path(path) if not parts: return current: MutableMapping[str, Any] = container for part in parts[:-1]: nested = current.get(part) if not isinstance(nested, MutableMapping): nested = {} current[part] = nested current = nested current[parts[-1]] = value def _delete_path(container: MutableMapping[str, Any], path: str) -> None: """Delete a possibly nested value from a mapping using dot notation.""" parts = _split_path(path) if not parts: return current: MutableMapping[str, Any] = container for part in parts[:-1]: nested = current.get(part) if not isinstance(nested, MutableMapping): return current = nested current.pop(parts[-1], None) def _increment_path(container: MutableMapping[str, Any], path: str, delta: float) -> float: """Increment a numeric value at a nested path and return the updated value.""" current_value = _get_path(container, path, 0.0) updated_value = _as_float(current_value, 0.0) + float(delta) _set_path(container, path, updated_value) return updated_value def _position_tuple(value: Any) -> Position | None: """Convert a position-like value into a numeric tuple.""" if value is None: return None try: array = np.asarray(value, dtype=float) except (TypeError, ValueError): return None if array.ndim != 1 or array.size == 0: return None return tuple(float(component) for component in array.tolist()) def _agent_position(agent: Agent) -> Position | None: """Return an agent's position as a numeric tuple if possible.""" return _position_tuple(getattr(agent, "position", None)) def _iter_world_agents(world: World) -> tuple[Agent, ...]: """Return all agent-like objects from a world.""" agents = getattr(world, "agents", ()) if isinstance(agents, Mapping): raw_agents: Iterable[Any] = agents.values() elif isinstance(agents, Iterable) and not isinstance(agents, (str, bytes)): raw_agents = agents else: raw_agents = () return tuple(agent for agent in raw_agents if hasattr(agent, "id")) def _find_agent(world: World, agent_id: str | None) -> Agent | None: """Find an agent by id.""" if agent_id is None: return None expected_id = str(agent_id) for candidate in _iter_world_agents(world): if _agent_id(candidate) == expected_id: return candidate return None def _structure_store( owner: Agent, *, location: StorageLocation | str, structures_key: str, ) -> MutableMapping[str, Any]: """Return the mutable structure store for an owner, creating it if needed.""" container = _container_for(owner, location) existing = _get_path(container, structures_key) if isinstance(existing, MutableMapping): return existing if isinstance(existing, Mapping): replacement = dict(existing) _set_path(container, structures_key, replacement) return replacement replacement: dict[str, Any] = {} _set_path(container, structures_key, replacement) return replacement def _existing_structure_store( owner: Agent, *, location: StorageLocation | str, structures_key: str, ) -> MutableMapping[str, Any] | None: """Return an existing mutable structure store, if one exists.""" container = _container_for(owner, location) existing = _get_path(container, structures_key) if isinstance(existing, MutableMapping): return existing if isinstance(existing, Mapping): replacement = dict(existing) _set_path(container, structures_key, replacement) return replacement return None def _find_structure( agent: Agent, world: World, *, structure_id: str, owner_agent_id: str | None, search_all_agents: bool, location: StorageLocation | str, structures_key: str, ) -> StructureHandle | None: """Find a mutable structure record by id.""" if not structure_id: return None normalized_location = _normalize_storage(location) if owner_agent_id is not None: owners = (_find_agent(world, owner_agent_id),) elif search_all_agents: owners = _iter_world_agents(world) else: owners = (agent,) for owner in owners: if owner is None: continue store = _existing_structure_store( owner, location=normalized_location, structures_key=structures_key, ) if store is None: continue raw_record = store.get(structure_id) if isinstance(raw_record, MutableMapping): return StructureHandle( owner=owner, structure_id=structure_id, record=raw_record, store=store, location=normalized_location, store_key=structures_key, ) if isinstance(raw_record, Mapping): replacement = dict(raw_record) store[structure_id] = replacement return StructureHandle( owner=owner, structure_id=structure_id, record=replacement, store=store, location=normalized_location, store_key=structures_key, ) return None def _record_status(record: Mapping[str, Any], status_key: str) -> str: """Return a structure status as a string.""" return str(record.get(status_key, "")) def _is_demolished(record: Mapping[str, Any], status_key: str) -> bool: """Return whether a structure record is marked as demolished.""" return _record_status(record, status_key) == StructureStatus.DEMOLISHED.value def _inventory( agent: Agent, *, location: StorageLocation | str, inventory_key: str, ) -> MutableMapping[str, Any]: """Return an inventory mapping from an agent state or memory path.""" container = _container_for(agent, location) existing = _get_path(container, inventory_key) if isinstance(existing, MutableMapping): return existing if isinstance(existing, Mapping): replacement = dict(existing) _set_path(container, inventory_key, replacement) return replacement replacement: dict[str, Any] = {} _set_path(container, inventory_key, replacement) return replacement def _has_materials( agent: Agent, material_cost: Mapping[str, float], *, inventory_location: StorageLocation | str, inventory_key: str, ) -> bool: """Return whether an agent has enough numeric materials for a cost.""" if not material_cost: return True items = _inventory(agent, location=inventory_location, inventory_key=inventory_key) for item_key, required in material_cost.items(): required_amount = max(0.0, float(required)) if required_amount <= 0: continue available_amount = _as_float(items.get(item_key), 0.0) if available_amount < required_amount: return False return True def _consume_materials( agent: Agent, material_cost: Mapping[str, float], *, inventory_location: StorageLocation | str, inventory_key: str, ) -> dict[str, float]: """Consume numeric materials from an inventory and return consumed amounts.""" consumed: dict[str, float] = {} if not material_cost: return consumed items = _inventory(agent, location=inventory_location, inventory_key=inventory_key) for item_key, required in material_cost.items(): required_amount = max(0.0, float(required)) if required_amount <= 0: continue available_amount = _as_float(items.get(item_key), 0.0) actual_amount = min(available_amount, required_amount) remaining = available_amount - actual_amount if remaining <= 0: items.pop(item_key, None) else: items[item_key] = remaining consumed[item_key] = actual_amount return consumed def _deposit_materials( agent: Agent, materials: Mapping[str, float], *, inventory_location: StorageLocation | str, inventory_key: str, fraction: float = 1.0, ) -> dict[str, float]: """Deposit numeric materials into an inventory and return deposited amounts.""" deposited: dict[str, float] = {} if not materials: return deposited items = _inventory(agent, location=inventory_location, inventory_key=inventory_key) safe_fraction = max(0.0, float(fraction)) for item_key, amount in materials.items(): deposit_amount = max(0.0, float(amount)) * safe_fraction if deposit_amount <= 0: continue current_amount = _as_float(items.get(item_key), 0.0) items[item_key] = current_amount + deposit_amount deposited[item_key] = deposit_amount return deposited def _has_energy(agent: Agent, energy_key: str, energy_cost: float) -> bool: """Return whether an agent has enough energy for a behavior.""" if energy_cost <= 0: return True return _as_float(_get_path(_agent_state(agent), energy_key), 0.0) >= energy_cost def _consume_energy(agent: Agent, energy_key: str, energy_cost: float) -> None: """Consume energy from an agent if a positive cost is configured.""" if energy_cost > 0: _increment_path(_agent_state(agent), energy_key, -float(energy_cost)) def _record_history( agent: Agent, *, history_key: str, record: Mapping[str, Any], max_history: int, ) -> None: """Append a bounded construction history record to agent memory.""" history = _ensure_list(_agent_memory(agent), history_key) _append_bounded(history, copy.deepcopy(dict(record)), max_history) def _generated_structure_id( agent: Agent, *, structure_type: str, store: Mapping[str, Any], step: int | None, ) -> str: """Generate a deterministic structure id.""" step_label = "unknown_step" if step is None else str(step) sequence_number = len(store) return f"{step_label}:{_agent_id(agent)}:{structure_type}:{sequence_number}" def _success( behavior: str, agent: Agent, *, target_ids: Sequence[str] = (), structure_ids: Sequence[str] = (), details: Mapping[str, Any] | None = None, world: World | None = None, ) -> dict[str, Any]: """Build a successful behavior outcome dictionary.""" return ConstructionOutcome( behavior=behavior, actor_id=_agent_id(agent), success=True, target_ids=tuple(str(target_id) for target_id in target_ids), structure_ids=tuple(str(structure_id) for structure_id in structure_ids), step=_world_step(world) if world is not None else None, details=details or {}, ).to_dict() def _failure( behavior: str, agent: Agent, reason: str, *, details: Mapping[str, Any] | None = None, world: World | None = None, ) -> dict[str, Any]: """Build a failed behavior outcome dictionary.""" payload: dict[str, Any] = {"reason": reason} if details: payload.update(details) logger.debug("Behavior %s failed for agent %s: %s", behavior, _agent_id(agent), reason) return ConstructionOutcome( behavior=behavior, actor_id=_agent_id(agent), success=False, step=_world_step(world) if world is not None else None, details=payload, ).to_dict() @dataclass class BuildBehavior(Behavior): """Create or progress a generic structure. The created structure is a plain mutable record stored in an owner's state or memory. This can represent a house, road, server cluster, habitat, ritual circle, farm shed, spacecraft module, research lab, or any other DSL-defined constructed asset. """ name: ClassVar[str] = "build" structure_id: str | None = None structure_type: str = "structure" owner_agent_id: str | None = None position: Sequence[float] | None = None blueprint: Mapping[str, Any] = field(default_factory=dict) metadata: Mapping[str, Any] = field(default_factory=dict) structures_location: StorageLocation | str = StorageLocation.MEMORY structures_key: str = "structures" inventory_location: StorageLocation | str = StorageLocation.STATE inventory_key: str = "inventory" material_cost: Mapping[str, float] = field(default_factory=dict) consume_materials: bool = True work_amount: float = 1.0 work_required: float = 1.0 progress_key: str = "progress" work_required_key: str = "work_required" status_key: str = "status" durability_key: str = "durability" max_durability_key: str = "max_durability" initial_durability: float = 1.0 capacity_key: str = "capacity" initial_capacity: float = 0.0 energy_key: str = "energy" energy_cost: float = 0.0 allow_rebuild_demolished: bool = False allow_existing_complete: bool = False history_memory_key: str = "construction_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can perform build work this step.""" if not _is_alive(agent): return False if self.work_amount <= 0 or self.work_required <= 0: return False owner = self._owner(agent, world) if owner is None: return False if not _has_energy(agent, self.energy_key, self.energy_cost): return False if self.consume_materials and not _has_materials( agent, self.material_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ): return False if self.structure_id is None: return True handle = _find_structure( agent, world, structure_id=self.structure_id, owner_agent_id=self.owner_agent_id, search_all_agents=False, location=self.structures_location, structures_key=self.structures_key, ) if handle is None: return True if _is_demolished(handle.record, self.status_key) and not self.allow_rebuild_demolished: return False progress = _as_float(handle.record.get(self.progress_key), 0.0) required = _as_float(handle.record.get(self.work_required_key), self.work_required) if progress >= required and not self.allow_existing_complete: return False return True def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Create or progress a structure record.""" if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) owner = self._owner(agent, world) if owner is None: return _failure(self.name, agent, "owner_not_available", world=world) step = _world_step(world) store = _structure_store( owner, location=self.structures_location, structures_key=self.structures_key, ) structure_id = self.structure_id or _generated_structure_id( agent, structure_type=self.structure_type, store=store, step=step, ) record = self._get_or_create_record( agent=agent, owner=owner, store=store, structure_id=structure_id, step=step, ) if _is_demolished(record, self.status_key) and self.allow_rebuild_demolished: record[self.status_key] = StructureStatus.ACTIVE.value record[self.progress_key] = 0.0 record[self.durability_key] = 0.0 consumed_materials: dict[str, float] = {} if self.consume_materials: consumed_materials = _consume_materials( agent, self.material_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ) _consume_energy(agent, self.energy_key, self.energy_cost) previous_progress = _as_float(record.get(self.progress_key), 0.0) required_work = max( 0.000001, _as_float(record.get(self.work_required_key), self.work_required), ) new_progress = min(required_work, previous_progress + float(self.work_amount)) record[self.progress_key] = new_progress record[self.work_required_key] = required_work record["updated_step"] = step completed = new_progress >= required_work if completed: record[self.status_key] = StructureStatus.COMPLETE.value record["completed_step"] = step record[self.durability_key] = max( _as_float(record.get(self.durability_key), 0.0), float(self.initial_durability), ) else: record[self.status_key] = StructureStatus.ACTIVE.value history_record = { "behavior": self.name, "structure_id": structure_id, "structure_type": self.structure_type, "owner_id": _agent_id(owner), "progress_before": previous_progress, "progress_after": new_progress, "work_required": required_work, "completed": completed, "materials": consumed_materials, "energy_cost": self.energy_cost, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) if _agent_id(owner) != _agent_id(agent): _record_history( owner, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s built structure %s for owner %s: %.3f/%.3f", _agent_id(agent), structure_id, _agent_id(owner), new_progress, required_work, ) return _success( self.name, agent, target_ids=(_agent_id(owner),), structure_ids=(structure_id,), details={ "structure_id": structure_id, "structure_type": self.structure_type, "owner_id": _agent_id(owner), "progress": new_progress, "work_required": required_work, "completed": completed, "status": record[self.status_key], "materials": consumed_materials, "energy_cost": self.energy_cost, }, world=world, ) def _owner(self, agent: Agent, world: World) -> Agent | None: """Return the owner receiving the structure.""" if self.owner_agent_id is None: return agent return _find_agent(world, self.owner_agent_id) def _get_or_create_record( self, *, agent: Agent, owner: Agent, store: MutableMapping[str, Any], structure_id: str, step: int | None, ) -> MutableMapping[str, Any]: """Return an existing structure record or create a new one.""" existing = store.get(structure_id) if isinstance(existing, MutableMapping): return existing if isinstance(existing, Mapping): replacement = dict(existing) store[structure_id] = replacement return replacement position = _position_tuple(self.position) or _agent_position(agent) record: dict[str, Any] = { "id": structure_id, "type": self.structure_type, "owner_id": _agent_id(owner), "created_by": _agent_id(agent), "position": position, self.status_key: StructureStatus.PLANNED.value, self.progress_key: 0.0, self.work_required_key: max(0.000001, float(self.work_required)), self.durability_key: 0.0, self.max_durability_key: max(0.0, float(self.initial_durability)), self.capacity_key: max(0.0, float(self.initial_capacity)), "blueprint": copy.deepcopy(dict(self.blueprint)), "metadata": copy.deepcopy(dict(self.metadata)), "created_step": step, "updated_step": step, } store[structure_id] = record return record @dataclass class RepairBehavior(Behavior): """Restore durability on a generic structure.""" name: ClassVar[str] = "repair" structure_id: str = "" owner_agent_id: str | None = None search_all_agents: bool = False structures_location: StorageLocation | str = StorageLocation.MEMORY structures_key: str = "structures" inventory_location: StorageLocation | str = StorageLocation.STATE inventory_key: str = "inventory" material_cost: Mapping[str, float] = field(default_factory=dict) consume_materials: bool = True repair_amount: float = 1.0 durability_key: str = "durability" max_durability_key: str = "max_durability" status_key: str = "status" energy_key: str = "energy" energy_cost: float = 0.0 restore_status_when_full: StructureStatus | str = StructureStatus.COMPLETE damaged_status: StructureStatus | str = StructureStatus.DAMAGED history_memory_key: str = "construction_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can repair the target structure.""" if not _is_alive(agent) or not self.structure_id or self.repair_amount <= 0: return False handle = self._handle(agent, world) if handle is None or _is_demolished(handle.record, self.status_key): return False current = _as_float(handle.record.get(self.durability_key), 0.0) maximum = _as_float(handle.record.get(self.max_durability_key), current) if maximum <= 0 or current >= maximum: return False if not _has_energy(agent, self.energy_key, self.energy_cost): return False if self.consume_materials and not _has_materials( agent, self.material_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ): return False return True def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Repair a structure by increasing its durability.""" handle = self._handle(agent, world) if handle is None: return _failure(self.name, agent, "structure_not_found", world=world) if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) record = handle.record step = _world_step(world) previous_durability = _as_float(record.get(self.durability_key), 0.0) maximum_durability = _as_float(record.get(self.max_durability_key), previous_durability) new_durability = min(maximum_durability, previous_durability + float(self.repair_amount)) consumed_materials: dict[str, float] = {} if self.consume_materials: consumed_materials = _consume_materials( agent, self.material_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ) _consume_energy(agent, self.energy_key, self.energy_cost) record[self.durability_key] = new_durability record["updated_step"] = step if new_durability >= maximum_durability: record[self.status_key] = str( self.restore_status_when_full.value if isinstance(self.restore_status_when_full, StructureStatus) else self.restore_status_when_full ) else: record[self.status_key] = str( self.damaged_status.value if isinstance(self.damaged_status, StructureStatus) else self.damaged_status ) history_record = { "behavior": self.name, "structure_id": handle.structure_id, "owner_id": _agent_id(handle.owner), "durability_before": previous_durability, "durability_after": new_durability, "max_durability": maximum_durability, "materials": consumed_materials, "energy_cost": self.energy_cost, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) if _agent_id(handle.owner) != _agent_id(agent): _record_history( handle.owner, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s repaired structure %s: %.3f -> %.3f", _agent_id(agent), handle.structure_id, previous_durability, new_durability, ) return _success( self.name, agent, target_ids=(_agent_id(handle.owner),), structure_ids=(handle.structure_id,), details={ "structure_id": handle.structure_id, "owner_id": _agent_id(handle.owner), "durability_before": previous_durability, "durability_after": new_durability, "max_durability": maximum_durability, "status": record[self.status_key], "materials": consumed_materials, "energy_cost": self.energy_cost, }, world=world, ) def _handle(self, agent: Agent, world: World) -> StructureHandle | None: """Return the target structure handle.""" return _find_structure( agent, world, structure_id=self.structure_id, owner_agent_id=self.owner_agent_id, search_all_agents=self.search_all_agents, location=self.structures_location, structures_key=self.structures_key, ) @dataclass class ExpandBehavior(Behavior): """Increase a generic numeric dimension of a structure. The default dimension is ``capacity``, but DSL authors can set ``dimension_key`` to anything numeric, such as size, throughput, population_limit, storage, bandwidth, housing, cargo_slots, mana_capacity, compute_units, or influence_radius. """ name: ClassVar[str] = "expand" structure_id: str = "" owner_agent_id: str | None = None search_all_agents: bool = False structures_location: StorageLocation | str = StorageLocation.MEMORY structures_key: str = "structures" inventory_location: StorageLocation | str = StorageLocation.STATE inventory_key: str = "inventory" material_cost: Mapping[str, float] = field(default_factory=dict) consume_materials: bool = True dimension_key: str = "capacity" expansion_amount: float = 1.0 max_dimension_value: float | None = None work_amount: float = 1.0 work_required: float = 1.0 expansion_progress_key: str = "expansion_progress" expansion_count_key: str = "expansion_count" status_key: str = "status" energy_key: str = "energy" energy_cost: float = 0.0 history_memory_key: str = "construction_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can expand the target structure.""" if ( not _is_alive(agent) or not self.structure_id or self.expansion_amount <= 0 or self.work_amount <= 0 or self.work_required <= 0 ): return False handle = self._handle(agent, world) if handle is None or _is_demolished(handle.record, self.status_key): return False current_dimension = _as_float(handle.record.get(self.dimension_key), 0.0) if self.max_dimension_value is not None and current_dimension >= self.max_dimension_value: return False if not _has_energy(agent, self.energy_key, self.energy_cost): return False if self.consume_materials and not _has_materials( agent, self.material_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ): return False return True def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Progress and possibly complete a structure expansion.""" handle = self._handle(agent, world) if handle is None: return _failure(self.name, agent, "structure_not_found", world=world) if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) record = handle.record step = _world_step(world) consumed_materials: dict[str, float] = {} if self.consume_materials: consumed_materials = _consume_materials( agent, self.material_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ) _consume_energy(agent, self.energy_key, self.energy_cost) previous_progress = _as_float(record.get(self.expansion_progress_key), 0.0) required = max(0.000001, float(self.work_required)) new_progress = previous_progress + float(self.work_amount) previous_dimension = _as_float(record.get(self.dimension_key), 0.0) expanded = new_progress >= required if expanded: requested_dimension = previous_dimension + float(self.expansion_amount) if self.max_dimension_value is not None: new_dimension = min(float(self.max_dimension_value), requested_dimension) else: new_dimension = requested_dimension record[self.dimension_key] = new_dimension record[self.expansion_progress_key] = 0.0 record[self.expansion_count_key] = _as_int(record.get(self.expansion_count_key), 0) + 1 record[self.status_key] = StructureStatus.EXPANDED.value record["last_expanded_step"] = step else: new_dimension = previous_dimension record[self.expansion_progress_key] = new_progress record["updated_step"] = step history_record = { "behavior": self.name, "structure_id": handle.structure_id, "owner_id": _agent_id(handle.owner), "dimension_key": self.dimension_key, "dimension_before": previous_dimension, "dimension_after": new_dimension, "progress_before": previous_progress, "progress_after": record[self.expansion_progress_key], "expanded": expanded, "materials": consumed_materials, "energy_cost": self.energy_cost, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) if _agent_id(handle.owner) != _agent_id(agent): _record_history( handle.owner, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s expanded structure %s dimension %s: %.3f -> %.3f", _agent_id(agent), handle.structure_id, self.dimension_key, previous_dimension, new_dimension, ) return _success( self.name, agent, target_ids=(_agent_id(handle.owner),), structure_ids=(handle.structure_id,), details={ "structure_id": handle.structure_id, "owner_id": _agent_id(handle.owner), "dimension_key": self.dimension_key, "dimension_before": previous_dimension, "dimension_after": new_dimension, "expanded": expanded, "progress": record[self.expansion_progress_key], "work_required": required, "materials": consumed_materials, "energy_cost": self.energy_cost, }, world=world, ) def _handle(self, agent: Agent, world: World) -> StructureHandle | None: """Return the target structure handle.""" return _find_structure( agent, world, structure_id=self.structure_id, owner_agent_id=self.owner_agent_id, search_all_agents=self.search_all_agents, location=self.structures_location, structures_key=self.structures_key, ) @dataclass class DemolishBehavior(Behavior): """Damage or demolish a generic structure. Demolition can either mark a structure as demolished, remove the record entirely, or apply partial durability damage. Optional salvage is deposited into the actor's inventory. """ name: ClassVar[str] = "demolish" structure_id: str = "" owner_agent_id: str | None = None search_all_agents: bool = False structures_location: StorageLocation | str = StorageLocation.MEMORY structures_key: str = "structures" inventory_location: StorageLocation | str = StorageLocation.STATE inventory_key: str = "inventory" damage_amount: float | None = None remove_record: bool = False salvage_yield: Mapping[str, float] = field(default_factory=dict) salvage_fraction: float = 1.0 demolition_cost: Mapping[str, float] = field(default_factory=dict) consume_demolition_cost: bool = True durability_key: str = "durability" status_key: str = "status" energy_key: str = "energy" energy_cost: float = 0.0 history_memory_key: str = "construction_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can demolish or damage the target structure.""" if not _is_alive(agent) or not self.structure_id: return False handle = self._handle(agent, world) if handle is None or _is_demolished(handle.record, self.status_key): return False if not _has_energy(agent, self.energy_key, self.energy_cost): return False if self.consume_demolition_cost and not _has_materials( agent, self.demolition_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ): return False return True def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Apply demolition damage and optionally salvage materials.""" handle = self._handle(agent, world) if handle is None: return _failure(self.name, agent, "structure_not_found", world=world) if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) record = handle.record step = _world_step(world) consumed_cost: dict[str, float] = {} if self.consume_demolition_cost: consumed_cost = _consume_materials( agent, self.demolition_cost, inventory_location=self.inventory_location, inventory_key=self.inventory_key, ) _consume_energy(agent, self.energy_key, self.energy_cost) previous_durability = _as_float(record.get(self.durability_key), 0.0) damage = previous_durability if self.damage_amount is None else max(0.0, float(self.damage_amount)) new_durability = max(0.0, previous_durability - damage) demolished = new_durability <= 0.0 or self.damage_amount is None record_snapshot = copy.deepcopy(dict(record)) record[self.durability_key] = new_durability record["updated_step"] = step if demolished: record[self.status_key] = StructureStatus.DEMOLISHED.value record["demolished_step"] = step else: record[self.status_key] = StructureStatus.DAMAGED.value salvage = _deposit_materials( agent, self.salvage_yield, inventory_location=self.inventory_location, inventory_key=self.inventory_key, fraction=self.salvage_fraction if demolished else 0.0, ) if demolished and self.remove_record: handle.store.pop(handle.structure_id, None) history_record = { "behavior": self.name, "structure_id": handle.structure_id, "owner_id": _agent_id(handle.owner), "durability_before": previous_durability, "durability_after": new_durability, "demolished": demolished, "removed": demolished and self.remove_record, "salvage": salvage, "cost": consumed_cost, "energy_cost": self.energy_cost, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) if _agent_id(handle.owner) != _agent_id(agent): _record_history( handle.owner, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s demolished structure %s: durability %.3f -> %.3f", _agent_id(agent), handle.structure_id, previous_durability, new_durability, ) return _success( self.name, agent, target_ids=(_agent_id(handle.owner),), structure_ids=(handle.structure_id,), details={ "structure_id": handle.structure_id, "owner_id": _agent_id(handle.owner), "durability_before": previous_durability, "durability_after": new_durability, "demolished": demolished, "removed": demolished and self.remove_record, "salvage": salvage, "cost": consumed_cost, "energy_cost": self.energy_cost, "record_before": record_snapshot, }, world=world, ) def _handle(self, agent: Agent, world: World) -> StructureHandle | None: """Return the target structure handle.""" return _find_structure( agent, world, structure_id=self.structure_id, owner_agent_id=self.owner_agent_id, search_all_agents=self.search_all_agents, location=self.structures_location, structures_key=self.structures_key, ) Build = BuildBehavior Repair = RepairBehavior Expand = ExpandBehavior Demolish = DemolishBehavior BEHAVIOR_REGISTRY: Mapping[str, type[Behavior]] = MappingProxyType( { BuildBehavior.name: BuildBehavior, RepairBehavior.name: RepairBehavior, ExpandBehavior.name: ExpandBehavior, DemolishBehavior.name: DemolishBehavior, } ) __all__ = [ "BEHAVIOR_REGISTRY", "Build", "BuildBehavior", "ConstructionOutcome", "Demolish", "DemolishBehavior", "Expand", "ExpandBehavior", "Position", "Repair", "RepairBehavior", "StorageLocation", "StructureHandle", "StructureStatus", ]