""" behaviors.attack ================ Generic conflict behaviors for WorldSmithAI. This module implements domain-agnostic conflict primitives: - AttackBehavior - HuntBehavior - 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. Model-friendly DSL examples --------------------------- Model-generated hunting: { "name": "hunt", "params": { "target": "deer", "effort": 1.0 } } Generic attack: { "name": "attack", "params": { "target": "wolf", "amount": 10.0, "target_state_key": "health", "mark_dead_on_threshold": true, "defeat_threshold": 0.0 } } Formal DSL-style attack: { "name": "attack", "params": { "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 } } Defense: { "name": "defend", "params": { "defense_state_key": "shield", "amount": 1.5, "max_state_value": 10.0 } } Raid: { "name": "raid", "params": { "target_selector": {"type": "colony"}, "radius": 3.0, "raid_assets": [ { "container": "inventory", "key": "fuel", "amount": 2.0 } ], "allow_partial": true } } """ from __future__ import annotations import logging from collections.abc import Mapping, Sequence from copy import deepcopy from dataclasses import dataclass, field, fields as dataclass_fields from numbers import Real from types import MappingProxyType from typing import Any, ClassVar, 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 = 10.0 DEFAULT_DEFENSE_AMOUNT: float = 1.0 DEFAULT_RAID_AMOUNT: float = 1.0 ZERO_TOLERANCE: float = 1.0e-12 _BEHAVIOR_BASE_FIELDS: set[str] = { "id", "behavior_id", "name", "parameters", "metadata", "enabled", "priority", "tags", "dsl_spec", } @dataclass(frozen=True, slots=True) class ConflictTargetCandidate: """Candidate target selected from the world.""" 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.""" 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).reshape(-1) 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.astype(np.float64, copy=True)) 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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. Direct fields are mirrored into ``self.parameters`` in ``__post_init__`` so model-generated shorthand such as ``{"target": "deer", "effort": 1.0}`` works with the same rich parameter system as formal DSL specs. """ target: str | None = None target_id: str | None = None target_agent_id: str | None = None victim_id: str | None = None target_type: str | None = None victim_type: str | None = None target_selector: Mapping[str, Any] | None = None selector: Mapping[str, Any] | None = None radius: float | None = None range: float | None = None interaction_radius: float | None = None attack_radius: float | None = None selection_strategy: str | None = None target_strategy: str | None = None selection_state_key: str | None = None amount: float | None = None damage: float | None = None damage_amount: float | None = None attack_amount: float | None = None power: float | None = None effect_amount: float | None = None effort: float | None = None target_state_key: str | None = None damage_state_key: str | None = None victim_state_key: str | None = None state_key: str | None = None target_state_operation: str | None = None target_state_updates: Mapping[str, Any] | None = None victim_state_updates: Mapping[str, Any] | None = None target_state_default: float | None = None min_target_state_value: float | None = None target_state_min: float | None = None max_target_state_value: float | None = None target_state_max: float | None = None mark_dead_on_threshold: bool | None = None defeat_threshold: float | None = None dead_threshold: float | None = None death_threshold: float | None = None actor_power_state_key: str | None = None power_state_key: str | None = None attack_power_state_key: str | None = None damage_multiplier: float | None = None effect_multiplier: float | None = None amount_multiplier: float | None = None actor_power_multiplier: float | None = None power_multiplier: float | None = None target_defense_state_key: str | None = None defense_state_key: str | None = None resistance_state_key: str | None = None defense_multiplier: float | None = None resistance_multiplier: float | None = None hit_probability: float | None = None success_probability: float | None = None probability: float | None = None cost_state_key: str | None = None state_cost_key: str | None = None budget_state_key: str | None = None cost: float | None = None action_cost: float | None = None base_cost: float | None = None cost_per_unit: float | None = None cost_per_amount: float | None = None cost_per_damage: float | None = None success_reward: float | None = None reward: float | None = None reward_per_unit: float | None = None reward_per_damage: float | None = None reward_per_amount: float | None = None reward_per_fraction: float | None = None partial_reward_multiplier: float | None = None failure_reward: float | None = None failure_penalty: float | None = None defense_amount: float | None = None shield_amount: float | None = None value: float | None = None target_self: bool | None = None mode: str | None = None output_state_key: str | None = None protection_state_key: str | None = None min_state_value: float | None = None defense_state_min: float | None = None max_state_value: float | None = None defense_state_max: float | None = None raid_assets: Any = None assets: Any = None loot: Any = None steal: Any = None item_key: str | None = None quantity: float | None = None item_container: str | None = None item_inventory_key: str | None = None inventory_key: str | None = None storage_key: str | None = None allow_partial: bool | None = None minimum_raid_fraction: float | None = None min_raid_fraction: float | None = None minimum_trade_fraction: float | None = None require_target_alive: bool | None = None def __post_init__(self) -> None: """Normalize base behavior state and mirror direct fields to parameters.""" super_post = getattr(super(), "__post_init__", None) if callable(super_post): super_post() raw_parameters = getattr(self, "parameters", {}) or {} if not isinstance(raw_parameters, Mapping): raise TypeError("ConflictBehavior.parameters must be a mapping.") parameters = dict(raw_parameters) for field_info in dataclass_fields(self): field_name = field_info.name if field_name in _BEHAVIOR_BASE_FIELDS: continue value = getattr(self, field_name, None) if value is None: continue parameters.setdefault(field_name, deepcopy(value)) self._set_parameters(parameters) def _set_parameters(self, parameters: Mapping[str, Any]) -> None: """Set parameters even when parent classes use slots.""" try: self.parameters = dict(parameters) except Exception: object.__setattr__(self, "parameters", dict(parameters)) def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """Check generic target and budget preconditions.""" 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, getattr(self, "id", self.__class__.__name__), ) return False def _select_target( self, *, agent: Agent, world: WorldProtocol, ) -> tuple[Agent, ConflictTargetCandidate] | None: """Select a target agent.""" 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.""" selector = self._target_selector() radius = self._radius() require_alive = self._optional_bool_parameter( "require_target_alive", default=True, ) agent_position = self._agent_position(agent) 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 other_position = self._agent_position(other) 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.""" selector_method = getattr(world, "select_agents", None) if callable(selector_method): try: selected = selector_method(selector) except TypeError: selected = selector_method(selector=selector) if isinstance(selected, Mapping): return list(selected.values()) if isinstance(selected, Sequence) and not isinstance(selected, (str, bytes)): return list(selected) return [] 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.""" 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.""" 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.""" del target multiplier = self._optional_finite_float_parameter( "damage_multiplier", aliases=("effect_multiplier", "amount_multiplier"), default=1.0, ) assert multiplier is not None effort = self._optional_non_negative_float_parameter( "effort", default=1.0, ) assert effort is not None raw_amount = base_amount * multiplier * effort 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.""" 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.""" 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.""" 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.""" 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.""" 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. Supports formal style: {"target_selector": {"type": "deer"}} and model-friendly shorthand: {"target": "deer"} """ 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"), ("target_agent_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]) raw_target = self.parameters.get("target") if isinstance(raw_target, str): target_text = raw_target.strip() if target_text and target_text.lower() not in { "agent", "agents", "resource", "resources", "target", "victim", }: normalized.setdefault("type", target_text) return normalized def _radius(self) -> float | None: """Return optional target interaction radius.""" 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.""" 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.""" 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. For model-friendly ``attack`` / ``hunt`` behaviors, this defaults to ``health`` when the DSL does not provide an explicit target state key. """ explicit = self._optional_string_parameter( "target_state_key", aliases=("damage_state_key", "victim_state_key", "state_key"), default=None, ) if explicit is not None: return explicit return str(self.parameters.get("default_target_state_key", "health")) def _target_state_operation(self) -> str: """Return operation used for target state mutation.""" 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.""" 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.""" new_value = value min_value = self._optional_finite_float_parameter( "min_target_state_value", aliases=("target_state_min",), default=0.0, ) 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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 _position_array(agent: Agent) -> NDArray[np.float64]: """Return agent position as a NumPy array, repairing tuple/list assignment.""" raw_position = getattr(agent, "position", None) if raw_position is None: position = np.zeros(2, dtype=np.float64) else: position = np.asarray(raw_position, dtype=np.float64).reshape(-1) if position.size == 0: position = np.zeros(2, dtype=np.float64) if not np.all(np.isfinite(position)): position = np.zeros(2, dtype=np.float64) if not isinstance(raw_position, np.ndarray): try: agent.set_position(position) except Exception: try: agent.position = position except Exception: logger.debug("Could not repair agent.position", exc_info=True) return position.astype(np.float64, copy=True) def _agent_position(self, agent: Agent) -> NDArray[np.float64]: """Return normalized agent position.""" return self._position_array(agent) @staticmethod def _numeric_state(agent: Agent, key: str, *, default: float) -> float: """Read a numeric agent 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.""" 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] elif key == "health": value = 100.0 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.""" 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.""" 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.""" 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.""" 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.""" rng = getattr(world, "rng", None) if isinstance(rng, np.random.Generator): return rng logger.debug("World does not expose a NumPy Generator as rng; using fallback RNG.") return np.random.default_rng(0) def _optional_bool_parameter(self, key: str, *, default: bool) -> bool: """Return optional 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" 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.""" def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """Check attack-specific preconditions.""" 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, getattr(self, "id", self.__class__.__name__), ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """Execute the attack.""" 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=self.name, 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=f"{self.__class__.__name__} missed target.", metadata=computation.to_metadata(), ) target.update_state(target_updates) if damage.marked_dead: target.mark_dead( reason=f"Threshold reached by behavior '{getattr(self, 'id', self.name)}'." ) computation = ConflictComputation( conflict_kind=self.name, 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=f"Agent performed {self.name}.", metadata=metadata, ) except Exception as exc: logger.exception( "%s failed for agent_id=%s behavior_id=%s", self.__class__.__name__, agent.id, getattr(self, "id", self.__class__.__name__), ) return self.failure( reward=self._failure_reward(), message=f"{self.__class__.__name__} raised an exception.", metadata={"error": repr(exc)}, ) def _has_attack_mutation_configured(self) -> bool: """Return whether the attack has a configured target mutation.""" return self._target_state_key() is not None or bool(self._target_state_updates()) @dataclass(slots=True) class HuntBehavior(AttackBehavior): """Compatibility alias for model-generated hunting behavior.""" @dataclass(slots=True) class DefendBehavior(ConflictBehavior): """Increase or set a defensive state value.""" def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """Check defend-specific preconditions.""" 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, getattr(self, "id", self.__class__.__name__), ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """Execute defense behavior.""" 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, getattr(self, "id", self.__class__.__name__), ) 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.""" 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.""" value = self._optional_string_parameter( "defense_state_key", aliases=("state_key", "output_state_key", "protection_state_key"), default=None, ) return value or "defense" def _defense_amount(self) -> float: """Return configured defense 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.""" 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.""" return self._optional_bool_parameter("target_self", default=True) def _apply_defense_clamps(self, value: float) -> float: """Apply optional defense state clamps.""" 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.""" def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """Check raid-specific preconditions.""" 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, getattr(self, "id", self.__class__.__name__), ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """Execute raid behavior.""" 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 '{getattr(self, 'id', self.name)}'." ) 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, getattr(self, "id", self.__class__.__name__), ) 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.""" 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.""" 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.""" 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.""" 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.""" return self._optional_bool_parameter("allow_partial", default=True) def _minimum_raid_fraction(self) -> float: """Return minimum partial raid fraction.""" 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 Hunt = HuntBehavior Defend = DefendBehavior Raid = RaidBehavior BEHAVIOR_REGISTRY: Mapping[str, type[Behavior]] = MappingProxyType( { "attack": AttackBehavior, "hunt": HuntBehavior, "defend": DefendBehavior, "raid": RaidBehavior, } ) __all__ = [ "AssetContainer", "Attack", "AttackBehavior", "BEHAVIOR_REGISTRY", "ConflictBehavior", "ConflictComputation", "ConflictTargetCandidate", "DamageComputation", "DEFAULT_ATTACK_AMOUNT", "DEFAULT_DEFENSE_AMOUNT", "DEFAULT_INVENTORY_KEY", "DEFAULT_RAID_AMOUNT", "Defend", "DefendBehavior", "Hunt", "HuntBehavior", "Raid", "RaidAsset", "RaidBehavior", "TargetSelectionStrategy", ]