""" behaviors.attack ================ Generic conflict behaviors for WorldSmithAI. This module implements domain-agnostic conflict primitives: - AttackBehavior - DefendBehavior - RaidBehavior These behaviors can model physical combat, economic rivalry, political attacks, reputation damage, market pressure, predation, raiding, sabotage, or defensive preparation. They do not assume hardcoded concepts such as health, armor, soldiers, wolves, weapons, money, or food. Example DSL-inspired attack --------------------------- { "id": "attack_enemy", "type": "attack", "parameters": { "target_selector": {"type": "bandit"}, "radius": 2.0, "target_state_key": "health", "amount": 2.0, "target_defense_state_key": "shield", "defense_multiplier": 0.5, "min_target_state_value": 0.0, "mark_dead_on_threshold": true, "defeat_threshold": 0.0 } } Example DSL-inspired defense ---------------------------- { "id": "raise_shield", "type": "defend", "parameters": { "defense_state_key": "shield", "amount": 1.5, "max_state_value": 10.0 } } Example DSL-inspired raid ------------------------- { "id": "raid_supplies", "type": "raid", "parameters": { "target_selector": {"type": "colony"}, "radius": 3.0, "raid_assets": [ { "container": "inventory", "key": "fuel", "amount": 2.0 } ], "allow_partial": true } } Minimal usage example --------------------- from core.agent import Agent from core.world import World from behaviors.attack import AttackBehavior world = World(seed=42) attacker = Agent( id="attacker_1", type="fighter", position=[0.0, 0.0], state={"stamina": 10.0}, behaviors={ "attack": AttackBehavior( id="attack", parameters={ "target_selector": {"type": "rival"}, "radius": 2.0, "target_state_key": "health", "amount": 3.0, "cost_state_key": "stamina", "cost": 1.0, "mark_dead_on_threshold": true, }, ) }, ) target = Agent( id="target_1", type="rival", position=[1.0, 0.0], state={"health": 5.0}, ) world.add_agent(attacker) world.add_agent(target) result = attacker.behaviors["attack"].execute(attacker, world) attacker.update_state(result.state_updates) """ from __future__ import annotations import logging from collections.abc import Mapping, Sequence from copy import deepcopy from dataclasses import dataclass, field from numbers import Real from typing import Any, Literal, TypeAlias import numpy as np from numpy.typing import NDArray from core.agent import Agent, WorldProtocol from core.behavior import Behavior, BehaviorResult logger = logging.getLogger(__name__) AssetContainer: TypeAlias = Literal["state", "inventory"] TargetSelectionStrategy: TypeAlias = Literal[ "nearest", "farthest", "random", "highest_state", "lowest_state", ] DEFAULT_INVENTORY_KEY: str = "inventory" DEFAULT_ATTACK_AMOUNT: float = 1.0 DEFAULT_DEFENSE_AMOUNT: float = 1.0 DEFAULT_RAID_AMOUNT: float = 1.0 ZERO_TOLERANCE: float = 1.0e-12 @dataclass(frozen=True, slots=True) class ConflictTargetCandidate: """ Candidate target selected from the world. Attributes ---------- id: Target agent id. type: Target agent type. position: Target position. distance: Distance from the acting agent. metadata: Additional target metadata for logs, metrics, or narration. """ id: str type: str position: NDArray[np.float64] distance: float metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize target candidate fields. Raises ------ ValueError If id, type, position, or distance are invalid. TypeError If metadata is invalid. """ if not isinstance(self.id, str) or not self.id.strip(): raise ValueError("ConflictTargetCandidate.id must be non-empty.") if not isinstance(self.type, str) or not self.type.strip(): raise ValueError("ConflictTargetCandidate.type must be non-empty.") position = np.asarray(self.position, dtype=np.float64) if position.ndim != 1: raise ValueError("ConflictTargetCandidate.position must be one-dimensional.") if position.size == 0: raise ValueError("ConflictTargetCandidate.position cannot be empty.") if not np.all(np.isfinite(position)): raise ValueError("ConflictTargetCandidate.position must contain finite values.") if not np.isfinite(self.distance): raise ValueError("ConflictTargetCandidate.distance must be finite.") if self.distance < 0.0: raise ValueError("ConflictTargetCandidate.distance cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("ConflictTargetCandidate.metadata must be a mapping.") object.__setattr__(self, "id", self.id.strip()) object.__setattr__(self, "type", self.type.strip()) object.__setattr__(self, "position", position) object.__setattr__(self, "distance", float(self.distance)) object.__setattr__(self, "metadata", deepcopy(dict(self.metadata))) def to_metadata(self) -> dict[str, Any]: """ Convert the target candidate into JSON-friendly metadata. Returns ------- dict[str, Any] Serializable target metadata. """ return { "id": self.id, "type": self.type, "position": self.position.tolist(), "distance": self.distance, "metadata": deepcopy(dict(self.metadata)), } @dataclass(frozen=True, slots=True) class RaidAsset: """ Asset transferred during a raid. Assets may live directly in an agent's state or inside an inventory mapping. Examples -------- State asset: RaidAsset(container="state", key="influence", amount=2.0) Inventory asset: RaidAsset(container="inventory", key="fuel", amount=3.0) Attributes ---------- key: State key or inventory item key. amount: Non-negative amount to transfer from target to actor. container: ``"state"`` or ``"inventory"``. inventory_key: Agent state key containing inventory when ``container="inventory"``. metadata: Optional structured metadata. """ key: str amount: float container: AssetContainer = "inventory" inventory_key: str = DEFAULT_INVENTORY_KEY metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize raid asset fields. Raises ------ ValueError If key, amount, container, or inventory_key are invalid. TypeError If metadata is invalid. """ if not isinstance(self.key, str) or not self.key.strip(): raise ValueError("RaidAsset.key must be a non-empty string.") if self.container not in {"state", "inventory"}: raise ValueError("RaidAsset.container must be 'state' or 'inventory'.") if not isinstance(self.inventory_key, str) or not self.inventory_key.strip(): raise ValueError("RaidAsset.inventory_key must be a non-empty string.") if isinstance(self.amount, bool) or not isinstance(self.amount, Real): raise TypeError("RaidAsset.amount must be numeric.") amount = float(self.amount) if not np.isfinite(amount): raise ValueError("RaidAsset.amount must be finite.") if amount < 0.0: raise ValueError("RaidAsset.amount cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("RaidAsset.metadata must be a mapping.") object.__setattr__(self, "key", self.key.strip()) object.__setattr__(self, "amount", amount) object.__setattr__(self, "inventory_key", self.inventory_key.strip()) object.__setattr__(self, "metadata", deepcopy(dict(self.metadata))) @classmethod def from_mapping( cls, raw_asset: Mapping[str, Any], *, default_container: AssetContainer = "inventory", default_inventory_key: str = DEFAULT_INVENTORY_KEY, ) -> RaidAsset: """ Build a RaidAsset from a DSL-style mapping. Parameters ---------- raw_asset: Raw asset mapping. default_container: Default storage container. default_inventory_key: Default inventory key. Returns ------- RaidAsset Normalized raid asset. Raises ------ TypeError If raw_asset is not a mapping. KeyError If required fields are missing. """ if not isinstance(raw_asset, Mapping): raise TypeError("Raid asset specification must be a mapping.") key = ( raw_asset.get("key") or raw_asset.get("item_key") or raw_asset.get("asset_key") or raw_asset.get("state_key") ) if key is None: raise KeyError("Raid asset requires 'key'.") amount = ( raw_asset.get("amount") if "amount" in raw_asset else raw_asset.get("quantity", raw_asset.get("value")) ) if amount is None: raise KeyError("Raid asset requires 'amount' or 'quantity'.") container = cls._normalize_container( raw_asset.get( "container", raw_asset.get("source", raw_asset.get("storage", default_container)), ) ) inventory_key = str( raw_asset.get( "inventory_key", raw_asset.get("storage_key", default_inventory_key), ) ) metadata = raw_asset.get("metadata", {}) if metadata is None: metadata = {} return cls( key=str(key), amount=amount, container=container, inventory_key=inventory_key, metadata=dict(metadata), ) def scaled(self, factor: float) -> RaidAsset: """ Return a copy with amount scaled by factor. Parameters ---------- factor: Non-negative scale factor. Returns ------- RaidAsset Scaled asset. """ if isinstance(factor, bool) or not isinstance(factor, Real): raise TypeError("factor must be numeric.") normalized_factor = float(factor) if not np.isfinite(normalized_factor): raise ValueError("factor must be finite.") if normalized_factor < 0.0: raise ValueError("factor cannot be negative.") return RaidAsset( key=self.key, amount=self.amount * normalized_factor, container=self.container, inventory_key=self.inventory_key, metadata=deepcopy(dict(self.metadata)), ) def to_metadata(self) -> dict[str, Any]: """ Convert the asset into JSON-friendly metadata. Returns ------- dict[str, Any] Serializable raid asset metadata. """ return { "key": self.key, "amount": self.amount, "container": self.container, "inventory_key": self.inventory_key, "metadata": deepcopy(dict(self.metadata)), } @staticmethod def _normalize_container(value: Any) -> AssetContainer: """ Normalize an asset container value. Parameters ---------- value: Candidate container value. Returns ------- AssetContainer ``"state"`` or ``"inventory"``. """ normalized = str(value).strip().lower() if normalized in {"state", "agent_state"}: return "state" if normalized in {"inventory", "inv"}: return "inventory" raise ValueError("Asset container must be 'state' or 'inventory'.") @dataclass(frozen=True, slots=True) class DamageComputation: """ Structured description of a damage or pressure effect. The word "damage" here means a generic negative pressure applied to a target state key. It can represent physical injury, morale loss, influence loss, market-share pressure, theory credibility loss, or any DSL-defined quantity. Attributes ---------- target_state_key: Target state key affected by the conflict effect. operation: State operation used: ``"decrement"``, ``"increment"``, ``"set"``, or ``"multiply"``. base_amount: Base configured amount before actor power and defense. raw_amount: Amount before defense mitigation. prevented_amount: Amount mitigated by target defense state. effective_amount: Amount actually applied to target state. previous_value: Previous target state value, if a target state key was used. new_value: New target state value, if a target state key was used. explicit_updates: Additional explicit target state updates. marked_dead: Whether the target was marked dead by threshold logic. hit: Whether the effect hit. """ target_state_key: str | None operation: str base_amount: float raw_amount: float prevented_amount: float effective_amount: float previous_value: float | None new_value: float | None explicit_updates: Mapping[str, Any] = field(default_factory=dict) marked_dead: bool = False hit: bool = True def __post_init__(self) -> None: """ Validate and normalize damage computation fields. Raises ------ ValueError If numeric fields are invalid. TypeError If explicit_updates or booleans are invalid. """ for label, value in { "base_amount": self.base_amount, "raw_amount": self.raw_amount, "prevented_amount": self.prevented_amount, "effective_amount": self.effective_amount, }.items(): if not np.isfinite(value): raise ValueError(f"{label} must be finite.") if self.previous_value is not None and not np.isfinite(self.previous_value): raise ValueError("previous_value must be finite when provided.") if self.new_value is not None and not np.isfinite(self.new_value): raise ValueError("new_value must be finite when provided.") if not isinstance(self.explicit_updates, Mapping): raise TypeError("explicit_updates must be a mapping.") if not isinstance(self.marked_dead, bool): raise TypeError("marked_dead must be a boolean.") if not isinstance(self.hit, bool): raise TypeError("hit must be a boolean.") object.__setattr__(self, "base_amount", float(self.base_amount)) object.__setattr__(self, "raw_amount", float(self.raw_amount)) object.__setattr__(self, "prevented_amount", float(self.prevented_amount)) object.__setattr__(self, "effective_amount", float(self.effective_amount)) object.__setattr__( self, "previous_value", None if self.previous_value is None else float(self.previous_value), ) object.__setattr__( self, "new_value", None if self.new_value is None else float(self.new_value), ) object.__setattr__(self, "explicit_updates", deepcopy(dict(self.explicit_updates))) def to_metadata(self) -> dict[str, Any]: """ Convert the computation into JSON-friendly metadata. Returns ------- dict[str, Any] Serializable damage metadata. """ return { "target_state_key": self.target_state_key, "operation": self.operation, "base_amount": self.base_amount, "raw_amount": self.raw_amount, "prevented_amount": self.prevented_amount, "effective_amount": self.effective_amount, "previous_value": self.previous_value, "new_value": self.new_value, "explicit_updates": deepcopy(dict(self.explicit_updates)), "marked_dead": self.marked_dead, "hit": self.hit, } @dataclass(frozen=True, slots=True) class ConflictComputation: """ Structured summary of a conflict behavior execution. Attributes ---------- conflict_kind: Conflict kind such as ``"attack"``, ``"defend"``, or ``"raid"``. actor_id: Acting agent id. actor_type: Acting agent type. target_id: Optional target id. target_type: Optional target type. target: Optional target metadata. actor_state_updates: State updates returned for the actor. target_state_updates: State updates applied directly to the target. damage: Optional damage computation metadata. transferred_assets: Assets transferred during a raid. partial: Whether execution was partial. metadata: Additional structured metadata. """ conflict_kind: str actor_id: str actor_type: str target_id: str | None = None target_type: str | None = None target: ConflictTargetCandidate | None = None actor_state_updates: Mapping[str, Any] = field(default_factory=dict) target_state_updates: Mapping[str, Any] = field(default_factory=dict) damage: DamageComputation | None = None transferred_assets: tuple[RaidAsset, ...] = field(default_factory=tuple) partial: bool = False metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize conflict computation fields. Raises ------ ValueError If identifiers are invalid. TypeError If nested data has invalid types. """ if not isinstance(self.conflict_kind, str) or not self.conflict_kind.strip(): raise ValueError("conflict_kind must be a non-empty string.") if not isinstance(self.actor_id, str) or not self.actor_id.strip(): raise ValueError("actor_id must be a non-empty string.") if not isinstance(self.actor_type, str) or not self.actor_type.strip(): raise ValueError("actor_type must be a non-empty string.") if self.target is not None and not isinstance(self.target, ConflictTargetCandidate): raise TypeError("target must be a ConflictTargetCandidate or None.") if not isinstance(self.actor_state_updates, Mapping): raise TypeError("actor_state_updates must be a mapping.") if not isinstance(self.target_state_updates, Mapping): raise TypeError("target_state_updates must be a mapping.") if self.damage is not None and not isinstance(self.damage, DamageComputation): raise TypeError("damage must be a DamageComputation or None.") if not isinstance(self.transferred_assets, tuple): object.__setattr__(self, "transferred_assets", tuple(self.transferred_assets)) for asset in self.transferred_assets: if not isinstance(asset, RaidAsset): raise TypeError("transferred_assets must contain RaidAsset objects.") if not isinstance(self.partial, bool): raise TypeError("partial must be a boolean.") if not isinstance(self.metadata, Mapping): raise TypeError("metadata must be a mapping.") object.__setattr__(self, "conflict_kind", self.conflict_kind.strip()) object.__setattr__(self, "actor_id", self.actor_id.strip()) object.__setattr__(self, "actor_type", self.actor_type.strip()) object.__setattr__(self, "actor_state_updates", deepcopy(dict(self.actor_state_updates))) object.__setattr__(self, "target_state_updates", deepcopy(dict(self.target_state_updates))) object.__setattr__(self, "metadata", deepcopy(dict(self.metadata))) def to_metadata(self) -> dict[str, Any]: """ Convert the computation into JSON-friendly metadata. Returns ------- dict[str, Any] Serializable conflict metadata. """ return { "conflict_kind": self.conflict_kind, "actor_id": self.actor_id, "actor_type": self.actor_type, "target_id": self.target_id, "target_type": self.target_type, "target": None if self.target is None else self.target.to_metadata(), "actor_state_updates": deepcopy(dict(self.actor_state_updates)), "target_state_updates": deepcopy(dict(self.target_state_updates)), "damage": None if self.damage is None else self.damage.to_metadata(), "transferred_assets": [ asset.to_metadata() for asset in self.transferred_assets ], "partial": self.partial, "metadata": deepcopy(dict(self.metadata)), } @dataclass(slots=True) class ConflictBehavior(Behavior): """ Base class for generic conflict behaviors. This class provides target selection, damage computation, action costs, random hit resolution, raid asset transfer helpers, and common parameter parsing. Concrete behaviors implement their own ``execute(...)`` methods. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check generic target and budget preconditions. Parameters ---------- agent: Acting agent. world: World containing possible targets. Returns ------- bool True if a target exists and cost can be paid. """ try: target = self._select_target(agent=agent, world=world) if target is None: return False return self._has_sufficient_action_budget( agent=agent, intensity=self._attack_amount(default=DEFAULT_ATTACK_AMOUNT), ) except Exception: logger.exception( "Conflict precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def _select_target( self, *, agent: Agent, world: WorldProtocol, ) -> tuple[Agent, ConflictTargetCandidate] | None: """ Select a target agent. Parameters ---------- agent: Acting agent. world: World containing agents. Returns ------- tuple[Agent, ConflictTargetCandidate] | None Selected target and metadata, or None. """ candidates = self._ordered_target_candidates(agent=agent, world=world) if not candidates: return None return candidates[0] def _ordered_target_candidates( self, *, agent: Agent, world: WorldProtocol, ) -> list[tuple[Agent, ConflictTargetCandidate]]: """ Return target candidates ordered by configured strategy. Parameters ---------- agent: Acting agent. world: World containing agents. Returns ------- list[tuple[Agent, ConflictTargetCandidate]] Ordered target candidates. """ selector = self._target_selector() radius = self._radius() require_alive = self._optional_bool_parameter( "require_target_alive", default=True, ) candidates: list[tuple[Agent, ConflictTargetCandidate]] = [] for other in self._agents_from_world(world=world, selector=selector): if not isinstance(other, Agent): continue if other.id == agent.id: continue if require_alive and not other.alive: continue if not self._matches_agent_selector(other, selector): continue if other.position.shape != agent.position.shape: continue distance = float(np.linalg.norm(other.position - agent.position)) if radius is not None and distance > radius: continue candidates.append( ( other, ConflictTargetCandidate( id=other.id, type=other.type, position=other.position.copy(), distance=distance, metadata={"class": other.__class__.__name__}, ), ) ) strategy = self._selection_strategy() if strategy == "nearest": candidates.sort(key=lambda item: (item[1].distance, item[1].id)) elif strategy == "farthest": candidates.sort(key=lambda item: (-item[1].distance, item[1].id)) elif strategy == "random": rng = self._rng(world) if len(candidates) > 1: indices = rng.permutation(len(candidates)) candidates = [candidates[int(index)] for index in indices] elif strategy in {"highest_state", "lowest_state"}: state_key = self._required_string_parameter( "selection_state_key", aliases=("target_state_rank_key", "rank_state_key"), ) reverse = strategy == "highest_state" candidates.sort( key=lambda item: ( -self._numeric_state(item[0], state_key, default=0.0) if reverse else self._numeric_state(item[0], state_key, default=0.0), item[1].distance, item[1].id, ) ) else: raise ValueError(f"Unsupported target selection strategy: {strategy}") return candidates def _agents_from_world( self, *, world: WorldProtocol, selector: Mapping[str, Any], ) -> list[Any]: """ Retrieve candidate agents from the world. Parameters ---------- world: World-like object. selector: Target selector. Returns ------- list[Any] Candidate agents. """ selector_method = getattr(world, "select_agents", None) if callable(selector_method): try: selected = selector_method(selector) except TypeError: selected = selector_method(selector=selector) return list(selected) agents = getattr(world, "agents", {}) if isinstance(agents, Mapping): return list(agents.values()) if isinstance(agents, Sequence) and not isinstance(agents, (str, bytes)): return list(agents) return [] def _matches_agent_selector( self, candidate: Agent, selector: Mapping[str, Any], ) -> bool: """ Return whether an agent matches a selector. Parameters ---------- candidate: Candidate agent. selector: Selector mapping. Returns ------- bool True if the candidate matches. """ if "id" in selector and not self._value_matches(candidate.id, selector["id"]): return False if "ids" in selector and not self._value_matches(candidate.id, selector["ids"]): return False if "type" in selector and not self._value_matches(candidate.type, selector["type"]): return False if "types" in selector and not self._value_matches(candidate.type, selector["types"]): return False if "alive" in selector and candidate.alive is not bool(selector["alive"]): return False if "state" in selector and not self._mapping_contains( candidate.state, selector["state"], ): return False if "state_min" in selector and not self._matches_numeric_thresholds( candidate.state, selector["state_min"], comparison="min", ): return False if "state_max" in selector and not self._matches_numeric_thresholds( candidate.state, selector["state_max"], comparison="max", ): return False return True def _compute_damage_updates( self, *, actor: Agent, target: Agent, base_amount: float, hit: bool, ) -> tuple[dict[str, Any], DamageComputation]: """ Compute target state updates for an attack-like effect. Parameters ---------- actor: Acting agent. target: Target agent. base_amount: Base configured effect amount. hit: Whether the attack hit. Returns ------- tuple[dict[str, Any], DamageComputation] Target state updates and damage computation metadata. """ target_state_key = self._target_state_key() explicit_updates = self._target_state_updates() normalized_base = self._normalize_non_negative_float( base_amount, "base_amount", ) if not hit: return {}, DamageComputation( target_state_key=target_state_key, operation=self._target_state_operation(), base_amount=normalized_base, raw_amount=0.0, prevented_amount=0.0, effective_amount=0.0, previous_value=None, new_value=None, explicit_updates={}, marked_dead=False, hit=False, ) raw_amount = self._raw_effect_amount( actor=actor, target=target, base_amount=normalized_base, ) prevented_amount = self._prevented_amount(target=target, raw_amount=raw_amount) effective_amount = max(0.0, raw_amount - prevented_amount) updates: dict[str, Any] = deepcopy(dict(explicit_updates)) previous_value: float | None = None new_value: float | None = None marked_dead = False operation = self._target_state_operation() if target_state_key is not None: previous_value = self._numeric_state_required_or_default( target, target_state_key, default_parameter="target_state_default", ) if operation == "decrement": new_value = previous_value - effective_amount elif operation == "increment": new_value = previous_value + effective_amount elif operation == "set": new_value = effective_amount elif operation == "multiply": new_value = previous_value * effective_amount else: raise ValueError( "target_state_operation must be one of: decrement, " "increment, set, multiply." ) new_value = self._apply_target_state_clamps(new_value) updates[target_state_key] = new_value if self._optional_bool_parameter("mark_dead_on_threshold", default=False): threshold = self._optional_finite_float_parameter( "defeat_threshold", aliases=("dead_threshold", "death_threshold"), default=0.0, ) assert threshold is not None marked_dead = new_value <= threshold if not updates: raise ValueError( "Conflict effect has no configured target mutation. Provide " "'target_state_key' or 'target_state_updates'." ) computation = DamageComputation( target_state_key=target_state_key, operation=operation, base_amount=normalized_base, raw_amount=raw_amount, prevented_amount=prevented_amount, effective_amount=effective_amount, previous_value=previous_value, new_value=new_value, explicit_updates=explicit_updates, marked_dead=marked_dead, hit=True, ) return updates, computation def _raw_effect_amount( self, *, actor: Agent, target: Agent, base_amount: float, ) -> float: """ Compute raw effect amount before mitigation. Parameters ---------- actor: Acting agent. target: Target agent. base_amount: Base amount. Returns ------- float Raw non-negative effect amount. """ multiplier = self._optional_finite_float_parameter( "damage_multiplier", aliases=("effect_multiplier", "amount_multiplier"), default=1.0, ) assert multiplier is not None raw_amount = base_amount * multiplier actor_power_key = self._optional_string_parameter( "actor_power_state_key", aliases=("power_state_key", "attack_power_state_key"), default=None, ) if actor_power_key is not None: actor_power = self._numeric_state(actor, actor_power_key, default=0.0) actor_power_multiplier = self._optional_finite_float_parameter( "actor_power_multiplier", aliases=("power_multiplier",), default=1.0, ) assert actor_power_multiplier is not None raw_amount += actor_power * actor_power_multiplier if raw_amount < 0.0: return 0.0 return raw_amount def _prevented_amount(self, *, target: Agent, raw_amount: float) -> float: """ Compute amount prevented by target defensive state. Parameters ---------- target: Target agent. raw_amount: Raw effect amount. Returns ------- float Non-negative prevented amount. """ defense_key = self._optional_string_parameter( "target_defense_state_key", aliases=("defense_state_key", "resistance_state_key"), default=None, ) if defense_key is None: return 0.0 defense_value = self._numeric_state(target, defense_key, default=0.0) defense_multiplier = self._optional_finite_float_parameter( "defense_multiplier", aliases=("resistance_multiplier",), default=1.0, ) assert defense_multiplier is not None prevented = defense_value * defense_multiplier if prevented <= 0.0: return 0.0 return min(raw_amount, prevented) def _build_cost_state_updates( self, *, agent: Agent, intensity: float, existing_updates: Mapping[str, Any], charge: bool = True, ) -> dict[str, Any]: """ Build actor state updates for optional action cost. Parameters ---------- agent: Acting agent. intensity: Action intensity used for cost-per-unit. existing_updates: Existing pending actor updates. charge: Whether to charge the cost. Returns ------- dict[str, Any] Cost state updates. """ if not charge: return {} key = self._cost_state_key() if key is None: return {} cost = self._action_cost(intensity) if cost <= ZERO_TOLERANCE: return {} if key in existing_updates: value = existing_updates[key] if isinstance(value, bool) or not isinstance(value, Real): raise TypeError( f"Cannot charge action cost against non-numeric pending key '{key}'." ) current = float(value) else: current = self._numeric_state(agent, key, default=0.0) return {key: current - cost} def _has_sufficient_action_budget(self, *, agent: Agent, intensity: float) -> bool: """ Return whether the agent can pay optional action cost. Parameters ---------- agent: Acting agent. intensity: Action intensity used for cost calculation. Returns ------- bool True if no cost exists or the agent can pay. """ key = self._cost_state_key() if key is None: return True cost = self._action_cost(intensity) if cost <= ZERO_TOLERANCE: return True current = self._numeric_state(agent, key, default=0.0) return current >= cost def _action_cost(self, intensity: float) -> float: """ Compute optional action cost. Parameters ---------- intensity: Action intensity. Returns ------- float Non-negative action cost. """ normalized_intensity = self._normalize_non_negative_float(intensity, "intensity") base_cost = self._optional_non_negative_float_parameter( "cost", aliases=("action_cost", "base_cost"), default=0.0, ) cost_per_unit = self._optional_non_negative_float_parameter( "cost_per_unit", aliases=("cost_per_amount", "cost_per_damage"), default=0.0, ) assert base_cost is not None assert cost_per_unit is not None return base_cost + cost_per_unit * normalized_intensity def _roll_hit(self, world: WorldProtocol) -> bool: """ Resolve probabilistic hit success. Parameters ---------- world: World providing optional RNG. Returns ------- bool True if the conflict effect hits. """ probability = self._optional_non_negative_float_parameter( "hit_probability", aliases=("success_probability", "probability"), default=1.0, ) assert probability is not None if probability > 1.0: raise ValueError("hit_probability cannot exceed 1.0.") if probability >= 1.0: return True if probability <= 0.0: return False return bool(self._rng(world).random() < probability) def _target_selector(self) -> dict[str, Any]: """ Return normalized target selector. Returns ------- dict[str, Any] Selector mapping. """ selector = self.parameters.get( "target_selector", self.parameters.get("selector", {}), ) if selector is None: selector = {} if not isinstance(selector, Mapping): raise TypeError("target_selector must be a mapping.") normalized = deepcopy(dict(selector)) self._validate_string_keys(normalized, "target_selector") for parameter_key, selector_key in ( ("target_id", "id"), ("victim_id", "id"), ("target_type", "type"), ("victim_type", "type"), ): if parameter_key in self.parameters and self.parameters[parameter_key] is not None: normalized.setdefault(selector_key, self.parameters[parameter_key]) return normalized def _radius(self) -> float | None: """ Return optional target interaction radius. Returns ------- float | None Non-negative radius or None. """ return self._optional_non_negative_float_parameter( "radius", aliases=("range", "interaction_radius", "attack_radius"), default=None, ) def _selection_strategy(self) -> TargetSelectionStrategy: """ Return target selection strategy. Returns ------- TargetSelectionStrategy Normalized selection strategy. """ raw_value = self.parameters.get( "selection_strategy", self.parameters.get("target_strategy", "nearest"), ) value = str(raw_value).strip().lower() if value in {"nearest", "closest"}: return "nearest" if value in {"farthest", "furthest"}: return "farthest" if value == "random": return "random" if value in {"highest_state", "max_state"}: return "highest_state" if value in {"lowest_state", "min_state"}: return "lowest_state" raise ValueError( "selection_strategy must be one of: nearest, farthest, random, " "highest_state, lowest_state." ) def _attack_amount(self, *, default: float) -> float: """ Return configured attack or pressure amount. Parameters ---------- default: Default amount. Returns ------- float Non-negative amount. """ for key in ( "amount", "damage", "damage_amount", "attack_amount", "power", "effect_amount", ): if key in self.parameters and self.parameters[key] is not None: return self._normalize_non_negative_float(self.parameters[key], key) return default def _target_state_key(self) -> str | None: """ Return optional target state key affected by attack. Returns ------- str | None Target state key, if configured. """ return self._optional_string_parameter( "target_state_key", aliases=("damage_state_key", "victim_state_key", "state_key"), default=None, ) def _target_state_operation(self) -> str: """ Return operation used for target state mutation. Returns ------- str Normalized operation name. """ operation = str( self.parameters.get("target_state_operation", "decrement") ).strip().lower() if operation in {"subtract", "reduce", "damage"}: return "decrement" if operation in {"add", "increase"}: return "increment" if operation in {"assign"}: return "set" if operation in {"scale"}: return "multiply" return operation def _target_state_updates(self) -> dict[str, Any]: """ Return explicit target state updates. Returns ------- dict[str, Any] Explicit updates. """ updates = self.parameters.get("target_state_updates") if updates is None: updates = self.parameters.get("victim_state_updates") if updates is None: return {} if not isinstance(updates, Mapping): raise TypeError("target_state_updates must be a mapping.") self._validate_string_keys(updates, "target_state_updates") return deepcopy(dict(updates)) def _apply_target_state_clamps(self, value: float) -> float: """ Apply optional target state clamps. Parameters ---------- value: Candidate target state value. Returns ------- float Clamped value. """ new_value = value min_value = self._optional_finite_float_parameter( "min_target_state_value", aliases=("target_state_min",), default=None, ) max_value = self._optional_finite_float_parameter( "max_target_state_value", aliases=("target_state_max",), default=None, ) if min_value is not None: new_value = max(new_value, min_value) if max_value is not None: new_value = min(new_value, max_value) return new_value def _cost_state_key(self) -> str | None: """ Return optional actor state key used to pay action cost. Returns ------- str | None Cost state key or None. """ return self._optional_string_parameter( "cost_state_key", aliases=("state_cost_key", "budget_state_key"), default=None, ) def _success_reward(self, *, intensity: float = 0.0, fraction: float = 1.0) -> float: """ Compute success reward. Parameters ---------- intensity: Conflict intensity. fraction: Executed fraction for partial raids. Returns ------- float Reward value. """ base_reward = self._optional_finite_float_parameter( "success_reward", aliases=("reward",), default=0.0, ) reward_per_unit = self._optional_finite_float_parameter( "reward_per_unit", aliases=("reward_per_damage", "reward_per_amount"), default=0.0, ) reward_per_fraction = self._optional_finite_float_parameter( "reward_per_fraction", aliases=("partial_reward_multiplier",), default=0.0, ) assert base_reward is not None assert reward_per_unit is not None assert reward_per_fraction is not None return base_reward + reward_per_unit * intensity + reward_per_fraction * fraction def _failure_reward(self) -> float: """ Return failure reward or penalty. Returns ------- float Failure reward value. """ reward = self._optional_finite_float_parameter( "failure_reward", aliases=("failure_penalty",), default=0.0, ) assert reward is not None return reward def _inventory_key(self) -> str: """ Return default inventory state key. Returns ------- str Inventory key. """ value = self._optional_string_parameter( "inventory_key", aliases=("storage_key",), default=DEFAULT_INVENTORY_KEY, ) assert value is not None return value def _read_asset_amount( self, *, agent: Agent, asset: RaidAsset, pending_updates: Mapping[str, Any], ) -> float: """ Read asset amount from state plus pending updates. Parameters ---------- agent: Agent whose asset is read. asset: Asset descriptor. pending_updates: Pending state updates. Returns ------- float Non-negative current amount. """ if asset.container == "state": value = pending_updates.get(asset.key, agent.state.get(asset.key, 0.0)) return self._normalize_asset_amount(value, f"agent.state['{asset.key}']") inventory = self._read_inventory( agent=agent, inventory_key=asset.inventory_key, pending_updates=pending_updates, ) value = inventory.get(asset.key, 0.0) return self._normalize_asset_amount( value, f"agent.state['{asset.inventory_key}']['{asset.key}']", ) def _write_asset_amount( self, *, agent: Agent, pending_updates: dict[str, Any], asset: RaidAsset, amount: float, ) -> None: """ Write asset amount to pending updates. Parameters ---------- agent: Agent whose state is updated. pending_updates: Pending state updates. asset: Asset descriptor. amount: New amount. """ normalized_amount = self._normalize_non_negative_float(amount, "amount") if normalized_amount <= ZERO_TOLERANCE: normalized_amount = 0.0 if asset.container == "state": pending_updates[asset.key] = normalized_amount return inventory = self._read_inventory( agent=agent, inventory_key=asset.inventory_key, pending_updates=pending_updates, ) updated_inventory = deepcopy(dict(inventory)) updated_inventory[asset.key] = normalized_amount pending_updates[asset.inventory_key] = updated_inventory def _add_asset( self, *, agent: Agent, pending_updates: dict[str, Any], asset: RaidAsset, amount: float, ) -> None: """ Add an asset amount into pending updates. Parameters ---------- agent: Agent receiving asset. pending_updates: Pending state updates. asset: Asset descriptor. amount: Amount to add. """ current = self._read_asset_amount( agent=agent, asset=asset, pending_updates=pending_updates, ) self._write_asset_amount( agent=agent, pending_updates=pending_updates, asset=asset, amount=current + amount, ) def _subtract_asset( self, *, agent: Agent, pending_updates: dict[str, Any], asset: RaidAsset, amount: float, ) -> None: """ Subtract an asset amount into pending updates. Parameters ---------- agent: Agent losing asset. pending_updates: Pending state updates. asset: Asset descriptor. amount: Amount to subtract. Raises ------ ValueError If the agent lacks the requested asset amount. """ current = self._read_asset_amount( agent=agent, asset=asset, pending_updates=pending_updates, ) new_amount = current - amount if new_amount < -ZERO_TOLERANCE: raise ValueError( f"Agent '{agent.id}' lacks sufficient asset '{asset.key}'. " f"Required {amount}, available {current}." ) self._write_asset_amount( agent=agent, pending_updates=pending_updates, asset=asset, amount=max(0.0, new_amount), ) def _read_inventory( self, *, agent: Agent, inventory_key: str, pending_updates: Mapping[str, Any], ) -> dict[str, Any]: """ Read an inventory mapping from state plus pending updates. Parameters ---------- agent: Agent whose inventory is read. inventory_key: State key containing inventory. pending_updates: Pending state updates. Returns ------- dict[str, Any] Inventory mapping. """ value = pending_updates.get(inventory_key, agent.state.get(inventory_key, {})) if value is None: return {} if not isinstance(value, Mapping): raise TypeError( f"Agent '{agent.id}' state key '{inventory_key}' must be a mapping." ) return deepcopy(dict(value)) def _available_asset_fraction( self, *, agent: Agent, assets: Sequence[RaidAsset], pending_updates: Mapping[str, Any], ) -> float: """ Compute fraction of requested assets available from an agent. Parameters ---------- agent: Agent whose assets are checked. assets: Requested assets. pending_updates: Pending state updates. Returns ------- float Available fraction. """ fractions: list[float] = [] for asset in assets: if asset.amount <= ZERO_TOLERANCE: continue available = self._read_asset_amount( agent=agent, asset=asset, pending_updates=pending_updates, ) fractions.append(available / asset.amount) if not fractions: return 1.0 return min(fractions) @staticmethod def _numeric_state(agent: Agent, key: str, *, default: float) -> float: """ Read a numeric agent state value. Parameters ---------- agent: Agent whose state is read. key: State key. default: Default value. Returns ------- float Numeric state value. """ value = agent.state.get(key, default) if isinstance(value, bool) or not isinstance(value, Real): return default normalized = float(value) if not np.isfinite(normalized): return default return normalized def _numeric_state_required_or_default( self, agent: Agent, key: str, *, default_parameter: str, ) -> float: """ Read required numeric state or parameter-provided default. Parameters ---------- agent: Agent whose state is read. key: State key. default_parameter: Parameter name containing a default value. Returns ------- float Numeric state value. Raises ------ KeyError If state key is absent and no default parameter exists. """ if key in agent.state: value = agent.state[key] elif default_parameter in self.parameters and self.parameters[default_parameter] is not None: value = self.parameters[default_parameter] else: raise KeyError( f"Target agent '{agent.id}' is missing required state key '{key}'. " f"Provide '{default_parameter}' to supply a default." ) if isinstance(value, bool) or not isinstance(value, Real): raise TypeError(f"Agent '{agent.id}' state key '{key}' must be numeric.") normalized = float(value) if not np.isfinite(normalized): raise ValueError(f"Agent '{agent.id}' state key '{key}' must be finite.") return normalized @staticmethod def _normalize_asset_amount(value: Any, label: str) -> float: """ Normalize an existing asset amount. Parameters ---------- value: Candidate amount. label: Human-readable label. Returns ------- float Non-negative finite amount. """ if isinstance(value, bool) or not isinstance(value, Real): raise TypeError(f"{label} must be numeric.") normalized = float(value) if not np.isfinite(normalized): raise ValueError(f"{label} must be finite.") if normalized < -ZERO_TOLERANCE: raise ValueError(f"{label} cannot be negative.") return max(0.0, normalized) @staticmethod def _matches_numeric_thresholds( mapping: Mapping[str, Any], thresholds: Any, *, comparison: str, ) -> bool: """ Return whether mapping values satisfy numeric thresholds. Parameters ---------- mapping: Candidate mapping. thresholds: Threshold mapping. comparison: ``"min"`` or ``"max"``. Returns ------- bool True if thresholds are satisfied. """ if not isinstance(thresholds, Mapping): raise TypeError("Numeric threshold selector must be a mapping.") for key, threshold in thresholds.items(): if not isinstance(key, str): raise TypeError("Numeric threshold keys must be strings.") if key not in mapping: return False value = mapping[key] if isinstance(value, bool) or not isinstance(value, Real): return False if isinstance(threshold, bool) or not isinstance(threshold, Real): raise TypeError("Numeric threshold values must be numeric.") normalized_value = float(value) normalized_threshold = float(threshold) if not np.isfinite(normalized_value) or not np.isfinite(normalized_threshold): return False if comparison == "min" and normalized_value < normalized_threshold: return False if comparison == "max" and normalized_value > normalized_threshold: return False return True @staticmethod def _value_matches(value: Any, expected: Any) -> bool: """ Return whether a value matches a scalar or collection filter. Parameters ---------- value: Candidate value. expected: Scalar or collection. Returns ------- bool True if matched. """ if isinstance(expected, (str, bytes)): return value == expected if isinstance(expected, set): return value in expected if isinstance(expected, frozenset): return value in expected if isinstance(expected, Sequence): return value in expected return value == expected @staticmethod def _mapping_contains(actual: Any, expected: Any) -> bool: """ Return whether actual mapping contains expected key/value pairs. Parameters ---------- actual: Candidate mapping. expected: Expected mapping. Returns ------- bool True if all expected pairs match. """ if not isinstance(actual, Mapping) or not isinstance(expected, Mapping): return False for key, expected_value in expected.items(): if key not in actual: return False if actual[key] != expected_value: return False return True @staticmethod def _rng(world: WorldProtocol) -> np.random.Generator: """ Return world RNG or fallback generator. Parameters ---------- world: World-like object. Returns ------- np.random.Generator Random generator. """ rng = getattr(world, "rng", None) if isinstance(rng, np.random.Generator): return rng logger.warning( "World does not expose a NumPy Generator as rng; using fallback RNG." ) return np.random.default_rng() def _optional_bool_parameter(self, key: str, *, default: bool) -> bool: """ Return optional boolean parameter. Parameters ---------- key: Parameter key. default: Default value. Returns ------- bool Boolean parameter. """ if key not in self.parameters or self.parameters[key] is None: return default value = self.parameters[key] if not isinstance(value, bool): raise TypeError(f"Parameter '{key}' must be a boolean.") return value def _optional_string_parameter( self, key: str, *, aliases: Sequence[str] = (), default: str | None, ) -> str | None: """ Return optional string parameter. Parameters ---------- key: Primary key. aliases: Alternative keys. default: Default value. Returns ------- str | None Normalized string. """ for candidate_key in (key, *aliases): if candidate_key in self.parameters and self.parameters[candidate_key] is not None: value = self.parameters[candidate_key] if not isinstance(value, str) or not value.strip(): raise ValueError( f"Parameter '{candidate_key}' must be a non-empty string." ) return value.strip() return default def _required_string_parameter( self, key: str, *, aliases: Sequence[str] = (), ) -> str: """ Return required string parameter. Parameters ---------- key: Primary key. aliases: Alternative keys. Returns ------- str Normalized string. """ value = self._optional_string_parameter( key, aliases=aliases, default=None, ) if value is None: expected = ", ".join((key, *aliases)) raise KeyError(f"Missing required string parameter. Expected one of: {expected}") return value def _optional_finite_float_parameter( self, key: str, *, aliases: Sequence[str] = (), default: float | None, ) -> float | None: """ Return optional finite float parameter. Parameters ---------- key: Primary key. aliases: Alternative keys. default: Default value. Returns ------- float | None Normalized finite float. """ for candidate_key in (key, *aliases): if candidate_key in self.parameters and self.parameters[candidate_key] is not None: return self._normalize_finite_float( self.parameters[candidate_key], candidate_key, ) return default def _optional_non_negative_float_parameter( self, key: str, *, aliases: Sequence[str] = (), default: float | None, ) -> float | None: """ Return optional non-negative float parameter. Parameters ---------- key: Primary key. aliases: Alternative keys. default: Default value. Returns ------- float | None Normalized non-negative float. """ value = self._optional_finite_float_parameter( key, aliases=aliases, default=default, ) if value is not None and value < 0.0: raise ValueError(f"Parameter '{key}' cannot be negative.") return value @staticmethod def _normalize_finite_float(value: Any, label: str) -> float: """ Normalize a finite float. Parameters ---------- value: Candidate value. label: Human-readable label. Returns ------- float Finite float. """ if isinstance(value, bool) or not isinstance(value, Real): raise TypeError(f"{label} must be numeric.") normalized = float(value) if not np.isfinite(normalized): raise ValueError(f"{label} must be finite.") return normalized def _normalize_non_negative_float(self, value: Any, label: str) -> float: """ Normalize a non-negative finite float. Parameters ---------- value: Candidate value. label: Human-readable label. Returns ------- float Non-negative finite float. """ normalized = self._normalize_finite_float(value, label) if normalized < 0.0: raise ValueError(f"{label} cannot be negative.") return normalized @staticmethod def _validate_string_keys(mapping: Mapping[str, Any], label: str) -> None: """ Validate mapping keys are strings. Parameters ---------- mapping: Mapping to validate. label: Human-readable label. Raises ------ TypeError If any key is not a string. """ for key in mapping: if not isinstance(key, str): raise TypeError(f"{label} keys must be strings.") @dataclass(slots=True) class AttackBehavior(ConflictBehavior): """ Apply a generic attack or pressure effect to a target agent. AttackBehavior mutates a target state key or explicit target state updates. It does not assume that the target has ``health``. The DSL must specify the affected state key. Common parameters ----------------- target_selector: Selector used to find target agents. radius: Optional maximum distance. target_state_key: Target state key to mutate. amount: Base amount to apply. target_state_operation: ``"decrement"`` by default. Also supports ``"increment"``, ``"set"``, and ``"multiply"``. target_defense_state_key: Optional target state key that mitigates the effect. mark_dead_on_threshold: Whether to call ``target.mark_dead(...)`` when the affected state falls below ``defeat_threshold``. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check attack-specific preconditions. Parameters ---------- agent: Acting agent. world: World containing targets. Returns ------- bool True if a target and target mutation are available. """ try: if not self._has_attack_mutation_configured(): return False if not self._has_sufficient_action_budget( agent=agent, intensity=self._attack_amount(default=DEFAULT_ATTACK_AMOUNT), ): return False return self._select_target(agent=agent, world=world) is not None except Exception: logger.exception( "Attack precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute the attack. Parameters ---------- agent: Acting agent. world: World containing possible targets. Returns ------- BehaviorResult Attack result. """ try: selected = self._select_target(agent=agent, world=world) if selected is None: return self.failure( reward=self._failure_reward(), message="AttackBehavior found no matching target.", metadata={ "target_selector": self._target_selector(), "radius": self._radius(), }, ) target, candidate = selected amount = self._attack_amount(default=DEFAULT_ATTACK_AMOUNT) if not self._has_sufficient_action_budget(agent=agent, intensity=amount): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget to attack.", metadata={ "required_cost": self._action_cost(amount), "cost_state_key": self._cost_state_key(), }, ) actor_updates = self._build_cost_state_updates( agent=agent, intensity=amount, existing_updates={}, charge=True, ) hit = self._roll_hit(world) target_updates, damage = self._compute_damage_updates( actor=agent, target=target, base_amount=amount, hit=hit, ) if not hit: computation = ConflictComputation( conflict_kind="attack", actor_id=agent.id, actor_type=agent.type, target_id=target.id, target_type=target.type, target=candidate, actor_state_updates=actor_updates, target_state_updates={}, damage=damage, partial=False, metadata={"hit": False}, ) return self.failure( reward=self._failure_reward(), state_updates=actor_updates, memory_updates={"conflict": computation.to_metadata()}, message="AttackBehavior missed target.", metadata=computation.to_metadata(), ) target.update_state(target_updates) if damage.marked_dead: target.mark_dead( reason=f"Threshold reached by attack behavior '{self.id}'." ) computation = ConflictComputation( conflict_kind="attack", actor_id=agent.id, actor_type=agent.type, target_id=target.id, target_type=target.type, target=candidate, actor_state_updates=actor_updates, target_state_updates=target_updates, damage=damage, partial=False, metadata={ "hit": True, "target_alive_after": target.alive, }, ) metadata = computation.to_metadata() return self.success( reward=self._success_reward(intensity=damage.effective_amount), state_updates=actor_updates, memory_updates={"conflict": metadata}, message="Agent attacked target.", metadata=metadata, ) except Exception as exc: logger.exception( "AttackBehavior failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message="AttackBehavior raised an exception.", metadata={"error": repr(exc)}, ) def _has_attack_mutation_configured(self) -> bool: """ Return whether the attack has a configured target mutation. Returns ------- bool True if target_state_key or target_state_updates is configured. """ return self._target_state_key() is not None or bool(self._target_state_updates()) @dataclass(slots=True) class DefendBehavior(ConflictBehavior): """ Increase or set a defensive state value. DefendBehavior can apply to the acting agent or to another selected agent. The affected key is configured by the DSL and can mean shield, armor, resilience, reputation, security, morale, or any other defensive concept. Common parameters ----------------- defense_state_key: State key to update. amount: Defense amount. target_self: Whether the actor defends itself. Defaults to True. target_selector: Used only when ``target_self`` is False. mode: ``"increment"`` by default. Also supports ``"set"`` and ``"multiply"``. max_state_value: Optional upper clamp. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check defend-specific preconditions. Parameters ---------- agent: Acting agent. world: World containing possible defended targets. Returns ------- bool True if defense can be applied. """ try: _ = self._defense_state_key() amount = self._defense_amount() if not self._has_sufficient_action_budget(agent=agent, intensity=amount): return False if self._target_self(): return True return self._select_target(agent=agent, world=world) is not None except Exception: logger.exception( "Defend precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute defense behavior. Parameters ---------- agent: Acting agent. world: World containing possible defended targets. Returns ------- BehaviorResult Defense result. """ try: amount = self._defense_amount() if not self._has_sufficient_action_budget(agent=agent, intensity=amount): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget to defend.", metadata={ "required_cost": self._action_cost(amount), "cost_state_key": self._cost_state_key(), }, ) if self._target_self(): recipient = agent candidate = None else: selected = self._select_target(agent=agent, world=world) if selected is None: return self.failure( reward=self._failure_reward(), message="DefendBehavior found no matching target to defend.", metadata={ "target_selector": self._target_selector(), "radius": self._radius(), }, ) recipient, candidate = selected defense_updates = self._defense_updates(recipient=recipient, amount=amount) actor_updates: dict[str, Any] = {} target_updates: dict[str, Any] = {} if recipient.id == agent.id: actor_updates.update(defense_updates) else: target_updates.update(defense_updates) cost_updates = self._build_cost_state_updates( agent=agent, intensity=amount, existing_updates=actor_updates, charge=True, ) actor_updates.update(cost_updates) if recipient.id != agent.id: recipient.update_state(target_updates) computation = ConflictComputation( conflict_kind="defend", actor_id=agent.id, actor_type=agent.type, target_id=recipient.id, target_type=recipient.type, target=candidate, actor_state_updates=actor_updates, target_state_updates=target_updates, damage=None, transferred_assets=tuple(), partial=False, metadata={ "defense_state_key": self._defense_state_key(), "amount": amount, "target_self": recipient.id == agent.id, "mode": self._defense_mode(), }, ) metadata = computation.to_metadata() return self.success( reward=self._success_reward(intensity=amount), state_updates=actor_updates, memory_updates={"conflict": metadata}, message="Agent performed defense behavior.", metadata=metadata, ) except Exception as exc: logger.exception( "DefendBehavior failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message="DefendBehavior raised an exception.", metadata={"error": repr(exc)}, ) def _defense_updates(self, *, recipient: Agent, amount: float) -> dict[str, Any]: """ Build state updates for defense. Parameters ---------- recipient: Agent receiving defense. amount: Defense amount. Returns ------- dict[str, Any] State updates. """ key = self._defense_state_key() mode = self._defense_mode() current = self._numeric_state(recipient, key, default=0.0) if mode == "increment": new_value = current + amount elif mode == "set": new_value = amount elif mode == "multiply": new_value = current * amount else: raise ValueError("Defend mode must be one of: increment, set, multiply.") new_value = self._apply_defense_clamps(new_value) return {key: new_value} def _defense_state_key(self) -> str: """ Return required defense state key. Returns ------- str Defense state key. """ return self._required_string_parameter( "defense_state_key", aliases=("state_key", "output_state_key", "protection_state_key"), ) def _defense_amount(self) -> float: """ Return configured defense amount. Returns ------- float Non-negative amount. """ for key in ("amount", "defense_amount", "shield_amount", "value"): if key in self.parameters and self.parameters[key] is not None: return self._normalize_non_negative_float(self.parameters[key], key) return DEFAULT_DEFENSE_AMOUNT def _defense_mode(self) -> str: """ Return defense state update mode. Returns ------- str Normalized mode. """ mode = str(self.parameters.get("mode", "increment")).strip().lower() if mode in {"add", "increase"}: return "increment" if mode in {"assign"}: return "set" if mode in {"scale"}: return "multiply" return mode def _target_self(self) -> bool: """ Return whether defense applies to the actor. Returns ------- bool True if actor defends itself. """ return self._optional_bool_parameter("target_self", default=True) def _apply_defense_clamps(self, value: float) -> float: """ Apply optional defense state clamps. Parameters ---------- value: Candidate state value. Returns ------- float Clamped value. """ new_value = value min_value = self._optional_finite_float_parameter( "min_state_value", aliases=("defense_state_min",), default=None, ) max_value = self._optional_finite_float_parameter( "max_state_value", aliases=("defense_state_max",), default=None, ) if min_value is not None: new_value = max(new_value, min_value) if max_value is not None: new_value = min(new_value, max_value) return new_value @dataclass(slots=True) class RaidBehavior(ConflictBehavior): """ Transfer assets from a target to the acting agent. RaidBehavior can optionally also apply an attack-like state effect to the target when ``target_state_key`` or ``target_state_updates`` is configured. Common parameters ----------------- target_selector: Selector used to find target agents. raid_assets: List of assets to transfer from target to actor. allow_partial: Whether the raid can scale down if target lacks assets. Defaults to True. minimum_raid_fraction: Minimum fraction required for partial raids. target_state_key: Optional target state key to mutate during raid. amount: Optional attack amount if raid also applies state pressure. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check raid-specific preconditions. Parameters ---------- agent: Acting agent. world: World containing possible targets. Returns ------- bool True if a target exists and raid has configured assets or damage. """ try: if not self._raid_assets() and not self._has_optional_damage_configured(): return False intensity = self._raid_cost_intensity() if not self._has_sufficient_action_budget(agent=agent, intensity=intensity): return False return self._select_target(agent=agent, world=world) is not None except Exception: logger.exception( "Raid precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute raid behavior. Parameters ---------- agent: Acting agent. world: World containing possible targets. Returns ------- BehaviorResult Raid result. """ try: selected = self._select_target(agent=agent, world=world) if selected is None: return self.failure( reward=self._failure_reward(), message="RaidBehavior found no matching target.", metadata={ "target_selector": self._target_selector(), "radius": self._radius(), }, ) target, candidate = selected requested_assets = self._raid_assets() intensity = self._raid_cost_intensity() if not self._has_sufficient_action_budget(agent=agent, intensity=intensity): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget to raid.", metadata={ "required_cost": self._action_cost(intensity), "cost_state_key": self._cost_state_key(), }, ) actor_updates: dict[str, Any] = {} target_updates: dict[str, Any] = {} actor_updates.update( self._build_cost_state_updates( agent=agent, intensity=intensity, existing_updates=actor_updates, charge=True, ) ) hit = self._roll_hit(world) damage: DamageComputation | None = None if self._has_optional_damage_configured(): damage_updates, damage = self._compute_damage_updates( actor=agent, target=target, base_amount=self._attack_amount(default=0.0), hit=hit, ) target_updates.update(damage_updates) if not hit: computation = ConflictComputation( conflict_kind="raid", actor_id=agent.id, actor_type=agent.type, target_id=target.id, target_type=target.type, target=candidate, actor_state_updates=actor_updates, target_state_updates={}, damage=damage, transferred_assets=tuple(), partial=False, metadata={"hit": False}, ) return self.failure( reward=self._failure_reward(), state_updates=actor_updates, memory_updates={"conflict": computation.to_metadata()}, message="RaidBehavior missed target.", metadata=computation.to_metadata(), ) raid_fraction = self._raid_fraction( target=target, assets=requested_assets, pending_target_updates=target_updates, ) if requested_assets and raid_fraction <= ZERO_TOLERANCE: computation = ConflictComputation( conflict_kind="raid", actor_id=agent.id, actor_type=agent.type, target_id=target.id, target_type=target.type, target=candidate, actor_state_updates=actor_updates, target_state_updates=target_updates, damage=damage, transferred_assets=tuple(), partial=False, metadata={"reason": "insufficient_target_assets"}, ) return self.failure( reward=self._failure_reward(), state_updates=actor_updates, memory_updates={"conflict": computation.to_metadata()}, message="RaidBehavior could not transfer requested assets.", metadata=computation.to_metadata(), ) executed_assets = tuple(asset.scaled(raid_fraction) for asset in requested_assets) for asset in executed_assets: self._subtract_asset( agent=target, pending_updates=target_updates, asset=asset, amount=asset.amount, ) self._add_asset( agent=agent, pending_updates=actor_updates, asset=asset, amount=asset.amount, ) if target_updates: target.update_state(target_updates) if damage is not None and damage.marked_dead: target.mark_dead( reason=f"Threshold reached by raid behavior '{self.id}'." ) partial = raid_fraction < 1.0 - ZERO_TOLERANCE computation = ConflictComputation( conflict_kind="raid", actor_id=agent.id, actor_type=agent.type, target_id=target.id, target_type=target.type, target=candidate, actor_state_updates=actor_updates, target_state_updates=target_updates, damage=damage, transferred_assets=executed_assets, partial=partial, metadata={ "hit": True, "raid_fraction": raid_fraction, "target_alive_after": target.alive, }, ) metadata = computation.to_metadata() effective_intensity = 0.0 if damage is None else damage.effective_amount return self.success( reward=self._success_reward( intensity=effective_intensity, fraction=raid_fraction, ), state_updates=actor_updates, memory_updates={"conflict": metadata}, message="Agent raided target.", metadata=metadata, ) except Exception as exc: logger.exception( "RaidBehavior failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message="RaidBehavior raised an exception.", metadata={"error": repr(exc)}, ) def _raid_assets(self) -> tuple[RaidAsset, ...]: """ Return raid assets configured by parameters. Returns ------- tuple[RaidAsset, ...] Assets requested from target. """ raw_value = None for key in ("raid_assets", "assets", "loot", "steal"): if key in self.parameters and self.parameters[key] is not None: raw_value = self.parameters[key] break if raw_value is None and "item_key" in self.parameters: raw_value = { "key": self.parameters["item_key"], "amount": self.parameters.get( "quantity", self.parameters.get("amount", DEFAULT_RAID_AMOUNT), ), "container": self.parameters.get("item_container", "inventory"), "inventory_key": self.parameters.get( "item_inventory_key", self._inventory_key(), ), } if raw_value is None: return tuple() if isinstance(raw_value, Mapping): raw_assets = [raw_value] elif isinstance(raw_value, Sequence) and not isinstance(raw_value, (str, bytes)): raw_assets = list(raw_value) else: raise TypeError("raid_assets must be a mapping or sequence of mappings.") return tuple( RaidAsset.from_mapping( raw_asset, default_container=RaidAsset._normalize_container( self.parameters.get("item_container", "inventory") ), default_inventory_key=self._inventory_key(), ) for raw_asset in raw_assets ) def _raid_fraction( self, *, target: Agent, assets: Sequence[RaidAsset], pending_target_updates: Mapping[str, Any], ) -> float: """ Compute executable fraction of requested raid assets. Parameters ---------- target: Target agent. assets: Requested assets. pending_target_updates: Pending target updates. Returns ------- float Executable raid fraction in [0, 1]. """ if not assets: return 1.0 available_fraction = min( 1.0, self._available_asset_fraction( agent=target, assets=assets, pending_updates=pending_target_updates, ), ) if not self._allow_partial_raid(): return 1.0 if available_fraction >= 1.0 - ZERO_TOLERANCE else 0.0 minimum_fraction = self._minimum_raid_fraction() if available_fraction < minimum_fraction: return 0.0 return max(0.0, min(1.0, available_fraction)) def _raid_cost_intensity(self) -> float: """ Return intensity used for optional raid cost. Returns ------- float Non-negative cost intensity. """ asset_total = sum(asset.amount for asset in self._raid_assets()) damage_amount = ( self._attack_amount(default=0.0) if self._has_optional_damage_configured() else 0.0 ) return asset_total + damage_amount def _has_optional_damage_configured(self) -> bool: """ Return whether raid also includes target state mutation. Returns ------- bool True if raid has attack-like target mutation configured. """ return self._target_state_key() is not None or bool(self._target_state_updates()) def _allow_partial_raid(self) -> bool: """ Return whether partial raids are allowed. Returns ------- bool True if partial transfer is allowed. """ return self._optional_bool_parameter("allow_partial", default=True) def _minimum_raid_fraction(self) -> float: """ Return minimum partial raid fraction. Returns ------- float Fraction in [0, 1]. """ value = self._optional_non_negative_float_parameter( "minimum_raid_fraction", aliases=("min_raid_fraction", "minimum_trade_fraction"), default=ZERO_TOLERANCE, ) assert value is not None if value > 1.0: raise ValueError("minimum_raid_fraction cannot exceed 1.0.") return value Attack = AttackBehavior Defend = DefendBehavior Raid = RaidBehavior __all__ = [ "AssetContainer", "Attack", "AttackBehavior", "ConflictBehavior", "ConflictComputation", "ConflictTargetCandidate", "DamageComputation", "DEFAULT_ATTACK_AMOUNT", "DEFAULT_DEFENSE_AMOUNT", "DEFAULT_INVENTORY_KEY", "DEFAULT_RAID_AMOUNT", "Defend", "DefendBehavior", "Raid", "RaidAsset", "RaidBehavior", "TargetSelectionStrategy", ]