""" behaviors.research ================== Generic knowledge-production behaviors for WorldSmithAI. This module implements domain-agnostic research behaviors: - StudyBehavior - LearnBehavior - PublishBehavior - PatentBehavior - InnovateBehavior These behaviors can represent scientific research, magical study, cultural learning, startup innovation, technological invention, institutional knowledge creation, or any DSL-defined knowledge-production process. The module does not assume any specific agent type, profession, resource type, legal system, academic system, or economy. Example DSL-inspired behaviors ------------------------------ Study from a nearby knowledge resource: { "id": "study_archive", "type": "study", "parameters": { "source": "resource", "resource_selector": {"type": "archive"}, "amount": 1.0, "radius": 3.0, "state_gain_key": "knowledge", "state_gain_multiplier": 1.0, "deplete_resource": false } } Learn from another agent: { "id": "learn_from_mentor", "type": "learn", "parameters": { "target_selector": {"type": "mentor"}, "source_state_key": "knowledge", "state_gain_key": "knowledge", "amount": 1.0, "radius": 5.0, "decrement_source": false } } Publish a discovery: { "id": "publish_paper", "type": "publish", "parameters": { "input_state_key": "knowledge", "amount": 2.0, "consume_input": true, "output_state_key": "reputation", "output_multiplier": 1.5 } } Patent an invention: { "id": "patent_invention", "type": "patent", "parameters": { "input_state_key": "knowledge", "amount": 3.0, "output_state_key": "patents", "output_multiplier": 1.0, "output_state_updates": { "reputation": 0.5 } } } Innovate using a recipe: { "id": "innovate_product", "type": "innovate", "parameters": { "inputs": [ {"container": "state", "key": "knowledge", "amount": 2.0}, {"container": "inventory", "key": "capital", "amount": 5.0} ], "outputs": [ {"container": "state", "key": "innovation", "amount": 1.0}, {"container": "inventory", "key": "prototype", "amount": 1.0} ], "allow_partial": false, "success_probability": 1.0 } } Minimal usage example --------------------- from core.agent import Agent from core.world import World from behaviors.research import StudyBehavior world = World(seed=42) agent = Agent( id="researcher_1", type="researcher", position=[0.0, 0.0], state={"knowledge": 0.0, "energy": 10.0}, behaviors={ "study": StudyBehavior( id="study", parameters={ "source": "self", "amount": 1.0, "state_gain_key": "knowledge", "cost_state_key": "energy", "cost": 0.25, }, ) }, ) world.add_agent(agent) result = world.step() """ 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 from core.resource import Resource, ResourceOperationResult logger = logging.getLogger(__name__) AssetContainer: TypeAlias = Literal["state", "inventory"] ResearchSourceKind: TypeAlias = Literal["self", "resource", "agent"] SelectionStrategy: TypeAlias = Literal[ "nearest", "farthest", "random", "largest", "smallest", "highest_state", "lowest_state", ] DEFAULT_INVENTORY_KEY: str = "inventory" DEFAULT_KNOWLEDGE_STATE_KEY: str = "knowledge" DEFAULT_REPUTATION_STATE_KEY: str = "reputation" DEFAULT_PATENT_STATE_KEY: str = "patents" DEFAULT_INNOVATION_STATE_KEY: str = "innovation" DEFAULT_RESEARCH_AMOUNT: float = 1.0 ZERO_TOLERANCE: float = 1.0e-12 @dataclass(frozen=True, slots=True) class ResearchAsset: """ Generic state or inventory asset used by research behaviors. Assets can represent knowledge, funding, prototypes, mana, citations, compute, influence, reputation, or any other DSL-defined quantity. Attributes ---------- key: State key or inventory item key. amount: Non-negative quantity. container: ``"state"`` or ``"inventory"``. inventory_key: Agent state key containing the inventory mapping when ``container="inventory"``. metadata: Optional structured metadata. """ key: str amount: float container: AssetContainer = "state" inventory_key: str = DEFAULT_INVENTORY_KEY metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize 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("ResearchAsset.key must be a non-empty string.") if self.container not in {"state", "inventory"}: raise ValueError("ResearchAsset.container must be 'state' or 'inventory'.") if not isinstance(self.inventory_key, str) or not self.inventory_key.strip(): raise ValueError("ResearchAsset.inventory_key must be non-empty.") if isinstance(self.amount, bool) or not isinstance(self.amount, Real): raise TypeError("ResearchAsset.amount must be numeric.") amount = float(self.amount) if not np.isfinite(amount): raise ValueError("ResearchAsset.amount must be finite.") if amount < 0.0: raise ValueError("ResearchAsset.amount cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("ResearchAsset.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 = "state", default_inventory_key: str = DEFAULT_INVENTORY_KEY, ) -> ResearchAsset: """ Create an asset from a DSL-style mapping. Parameters ---------- raw_asset: Raw asset mapping. default_container: Default asset container. default_inventory_key: Default inventory key. Returns ------- ResearchAsset Normalized asset. Raises ------ TypeError If raw_asset is not a mapping. KeyError If required fields are missing. """ if not isinstance(raw_asset, Mapping): raise TypeError("Research asset specification must be a mapping.") key = ( raw_asset.get("key") or raw_asset.get("state_key") or raw_asset.get("item_key") or raw_asset.get("asset_key") ) if key is None: raise KeyError("Research 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("Research asset requires 'amount' or 'quantity'.") container = cls.normalize_container( raw_asset.get( "container", raw_asset.get("storage", raw_asset.get("source", 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) -> ResearchAsset: """ Return a copy of this asset scaled by a non-negative factor. Parameters ---------- factor: Non-negative scale factor. Returns ------- ResearchAsset Scaled asset. """ if isinstance(factor, bool) or not isinstance(factor, Real): raise TypeError("factor must be numeric.") normalized = float(factor) if not np.isfinite(normalized): raise ValueError("factor must be finite.") if normalized < 0.0: raise ValueError("factor cannot be negative.") return ResearchAsset( key=self.key, amount=self.amount * normalized, container=self.container, inventory_key=self.inventory_key, metadata=deepcopy(dict(self.metadata)), ) def to_metadata(self) -> dict[str, Any]: """ Convert the asset to JSON-friendly metadata. Returns ------- dict[str, Any] Serializable asset representation. """ 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 a 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 ResearchCandidate: """ Candidate knowledge source selected from the world. Attributes ---------- kind: ``"resource"`` or ``"agent"``. id: Source id. type: Source type. position: Source position. distance: Distance from acting agent. amount: Optional available amount associated with the source. metadata: Additional structured metadata. """ kind: Literal["resource", "agent"] id: str type: str position: NDArray[np.float64] distance: float amount: float | None = None metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize candidate fields. Raises ------ ValueError If fields are invalid. TypeError If metadata is invalid. """ if self.kind not in {"resource", "agent"}: raise ValueError("ResearchCandidate.kind must be 'resource' or 'agent'.") if not isinstance(self.id, str) or not self.id.strip(): raise ValueError("ResearchCandidate.id must be non-empty.") if not isinstance(self.type, str) or not self.type.strip(): raise ValueError("ResearchCandidate.type must be non-empty.") position = np.asarray(self.position, dtype=np.float64) if position.ndim != 1: raise ValueError("ResearchCandidate.position must be one-dimensional.") if position.size == 0: raise ValueError("ResearchCandidate.position cannot be empty.") if not np.all(np.isfinite(position)): raise ValueError("ResearchCandidate.position must contain finite values.") if not np.isfinite(self.distance): raise ValueError("ResearchCandidate.distance must be finite.") if self.distance < 0.0: raise ValueError("ResearchCandidate.distance cannot be negative.") if self.amount is not None: if not np.isfinite(self.amount): raise ValueError("ResearchCandidate.amount must be finite.") if self.amount < 0.0: raise ValueError("ResearchCandidate.amount cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("ResearchCandidate.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, "amount", None if self.amount is None else float(self.amount), ) object.__setattr__(self, "metadata", deepcopy(dict(self.metadata))) def to_metadata(self) -> dict[str, Any]: """ Convert the candidate to JSON-friendly metadata. Returns ------- dict[str, Any] Serializable candidate metadata. """ return { "kind": self.kind, "id": self.id, "type": self.type, "position": self.position.tolist(), "distance": self.distance, "amount": self.amount, "metadata": deepcopy(dict(self.metadata)), } @dataclass(frozen=True, slots=True) class ResearchComputation: """ Structured summary of a research behavior execution. Attributes ---------- research_kind: Behavior kind: ``"study"``, ``"learn"``, ``"publish"``, ``"patent"``, or ``"innovate"``. actor_id: Acting agent id. actor_type: Acting agent type. requested_amount: Requested amount or intensity. actual_amount: Actual completed amount or intensity. source: Optional selected source candidate. actor_state_updates: Updates returned for the actor. target_state_updates: Updates directly applied to another agent, if any. consumed_assets: Assets consumed from the actor. produced_assets: Assets produced for the actor. resource_results: Resource operation results. created_resources: World resources created by the behavior. partial: Whether execution was partial. metadata: Additional structured metadata. """ research_kind: str actor_id: str actor_type: str requested_amount: float actual_amount: float source: ResearchCandidate | None = None actor_state_updates: Mapping[str, Any] = field(default_factory=dict) target_state_updates: Mapping[str, Any] = field(default_factory=dict) consumed_assets: tuple[ResearchAsset, ...] = field(default_factory=tuple) produced_assets: tuple[ResearchAsset, ...] = field(default_factory=tuple) resource_results: tuple[ResourceOperationResult, ...] = field(default_factory=tuple) created_resources: tuple[dict[str, Any], ...] = field(default_factory=tuple) partial: bool = False metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize computation fields. Raises ------ ValueError If identifiers or amounts are invalid. TypeError If nested fields are invalid. """ if not isinstance(self.research_kind, str) or not self.research_kind.strip(): raise ValueError("research_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.") for label, value in { "requested_amount": self.requested_amount, "actual_amount": self.actual_amount, }.items(): if not np.isfinite(value): raise ValueError(f"{label} must be finite.") if value < 0.0: raise ValueError(f"{label} cannot be negative.") if self.source is not None and not isinstance(self.source, ResearchCandidate): raise TypeError("source must be a ResearchCandidate 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 not isinstance(self.consumed_assets, tuple): object.__setattr__(self, "consumed_assets", tuple(self.consumed_assets)) if not isinstance(self.produced_assets, tuple): object.__setattr__(self, "produced_assets", tuple(self.produced_assets)) if not isinstance(self.resource_results, tuple): object.__setattr__(self, "resource_results", tuple(self.resource_results)) if not isinstance(self.created_resources, tuple): object.__setattr__(self, "created_resources", tuple(self.created_resources)) for asset in self.consumed_assets: if not isinstance(asset, ResearchAsset): raise TypeError("consumed_assets must contain ResearchAsset objects.") for asset in self.produced_assets: if not isinstance(asset, ResearchAsset): raise TypeError("produced_assets must contain ResearchAsset objects.") for result in self.resource_results: if not isinstance(result, ResourceOperationResult): raise TypeError( "resource_results must contain ResourceOperationResult 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, "research_kind", self.research_kind.strip()) object.__setattr__(self, "actor_id", self.actor_id.strip()) object.__setattr__(self, "actor_type", self.actor_type.strip()) object.__setattr__(self, "requested_amount", float(self.requested_amount)) object.__setattr__(self, "actual_amount", float(self.actual_amount)) 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, "created_resources", tuple(deepcopy(list(self.created_resources))), ) 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 research computation metadata. """ return { "research_kind": self.research_kind, "actor_id": self.actor_id, "actor_type": self.actor_type, "requested_amount": self.requested_amount, "actual_amount": self.actual_amount, "source": None if self.source is None else self.source.to_metadata(), "actor_state_updates": deepcopy(dict(self.actor_state_updates)), "target_state_updates": deepcopy(dict(self.target_state_updates)), "consumed_assets": [ asset.to_metadata() for asset in self.consumed_assets ], "produced_assets": [ asset.to_metadata() for asset in self.produced_assets ], "resource_results": [ result.to_dict() for result in self.resource_results ], "created_resources": deepcopy(list(self.created_resources)), "partial": self.partial, "metadata": deepcopy(dict(self.metadata)), } @dataclass(slots=True) class ResearchBehavior(Behavior): """ Base class for generic research and knowledge-production behaviors. This class provides shared logic for: - resource selection - agent source selection - asset consumption and production - numeric state updates - optional action costs - optional world resource creation - deterministic randomness through ``world.rng`` """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check generic action budget preconditions. Parameters ---------- agent: Acting agent. world: World in which the behavior would execute. Returns ------- bool True if the agent can pay the default action cost. """ try: return self._has_sufficient_action_budget( agent=agent, intensity=self._amount(default=DEFAULT_RESEARCH_AMOUNT), ) except Exception: logger.exception( "Research precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def _select_resource( self, *, agent: Agent, world: WorldProtocol, selector: Mapping[str, Any] | None = None, ) -> tuple[Resource, ResearchCandidate] | None: """ Select a resource source from the world. Parameters ---------- agent: Acting agent. world: World containing resources. selector: Optional resource selector override. Returns ------- tuple[Resource, ResearchCandidate] | None Selected resource and metadata, or None. """ resource_selector = self._resource_selector() if selector is None else dict(selector) radius = self._radius() resources = self._resources_from_world(world=world, selector=resource_selector) candidates: list[tuple[Resource, ResearchCandidate]] = [] for resource in resources: if not isinstance(resource, Resource): continue if not self._matches_resource_selector(resource, resource_selector): continue if resource.position.shape != agent.position.shape: continue if self._optional_bool_parameter("ignore_empty_resources", default=True): if resource.amount <= ZERO_TOLERANCE: continue distance = float(np.linalg.norm(resource.position - agent.position)) if radius is not None and distance > radius: continue candidates.append( ( resource, ResearchCandidate( kind="resource", id=resource.id, type=resource.type, position=resource.position.copy(), distance=distance, amount=resource.amount, metadata=deepcopy(resource.metadata), ), ) ) if not candidates: return None strategy = self._selection_strategy(default="nearest") if strategy == "nearest": return min(candidates, key=lambda item: (item[1].distance, item[1].id)) if strategy == "farthest": return max(candidates, key=lambda item: (item[1].distance, item[1].id)) if strategy == "largest": return max(candidates, key=lambda item: (item[1].amount or 0.0, item[1].id)) if strategy == "smallest": return min(candidates, key=lambda item: (item[1].amount or 0.0, item[1].id)) if strategy == "random": index = int(self._rng(world).integers(0, len(candidates))) return candidates[index] raise ValueError(f"Unsupported resource selection strategy: {strategy}") def _select_agent_source( self, *, agent: Agent, world: WorldProtocol, selector: Mapping[str, Any] | None = None, ) -> tuple[Agent, ResearchCandidate] | None: """ Select another agent as a learning source. Parameters ---------- agent: Acting agent. world: World containing agents. selector: Optional selector override. Returns ------- tuple[Agent, ResearchCandidate] | None Selected source agent and metadata, or None. """ agent_selector = self._target_selector() if selector is None else dict(selector) radius = self._radius() require_alive = self._optional_bool_parameter( "require_target_alive", default=True, ) candidates: list[tuple[Agent, ResearchCandidate]] = [] for other in self._agents_from_world(world=world, selector=agent_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, agent_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, ResearchCandidate( kind="agent", id=other.id, type=other.type, position=other.position.copy(), distance=distance, amount=None, metadata={"class": other.__class__.__name__}, ), ) ) if not candidates: return None strategy = self._selection_strategy(default="nearest") if strategy == "nearest": return min(candidates, key=lambda item: (item[1].distance, item[1].id)) if strategy == "farthest": return max(candidates, key=lambda item: (item[1].distance, item[1].id)) if strategy == "random": index = int(self._rng(world).integers(0, len(candidates))) return candidates[index] if strategy in {"highest_state", "lowest_state"}: state_key = self._required_string_parameter( "selection_state_key", aliases=("source_state_rank_key", "rank_state_key"), ) reverse = strategy == "highest_state" sorted_candidates = sorted( candidates, key=lambda item: ( -self._numeric_agent_state(item[0], state_key, default=0.0) if reverse else self._numeric_agent_state(item[0], state_key, default=0.0), item[1].distance, item[1].id, ), ) return sorted_candidates[0] raise ValueError(f"Unsupported agent selection strategy: {strategy}") def _resources_from_world( self, *, world: WorldProtocol, selector: Mapping[str, Any], ) -> list[Any]: """ Retrieve resource candidates from the world. Parameters ---------- world: World-like object. selector: Resource selector. Returns ------- list[Any] Candidate resources. """ selector_method = getattr(world, "select_resources", None) if callable(selector_method): try: selected = selector_method(selector) except TypeError: selected = selector_method(selector=selector) return list(selected) resources = getattr(world, "resources", {}) if isinstance(resources, Mapping): return list(resources.values()) if isinstance(resources, Sequence) and not isinstance(resources, (str, bytes)): return list(resources) return [] def _agents_from_world( self, *, world: WorldProtocol, selector: Mapping[str, Any], ) -> list[Any]: """ Retrieve agent candidates from the world. Parameters ---------- world: World-like object. selector: Agent 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_resource_selector( self, resource: Resource, selector: Mapping[str, Any], ) -> bool: """ Return whether a resource matches a selector. Parameters ---------- resource: Candidate resource. selector: Selector mapping. Returns ------- bool True if the resource matches. """ if "id" in selector and not self._value_matches(resource.id, selector["id"]): return False if "ids" in selector and not self._value_matches(resource.id, selector["ids"]): return False if "type" in selector and not self._value_matches(resource.type, selector["type"]): return False if "types" in selector and not self._value_matches(resource.type, selector["types"]): return False if "metadata" in selector and not self._mapping_contains( resource.metadata, selector["metadata"], ): return False min_amount = selector.get("min_amount", selector.get("amount_min")) max_amount = selector.get("max_amount", selector.get("amount_max")) if min_amount is not None: if resource.amount < self._normalize_finite_float(min_amount, "min_amount"): return False if max_amount is not None: if resource.amount > self._normalize_finite_float(max_amount, "max_amount"): return False return True def _matches_agent_selector( self, candidate: Agent, selector: Mapping[str, Any], ) -> bool: """ Return whether an agent matches a selector. Parameters ---------- candidate: Candidate source agent. selector: Selector mapping. Returns ------- bool True if matched. """ 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 _resource_selector(self) -> dict[str, Any]: """ Build a resource selector from parameters. Returns ------- dict[str, Any] Selector mapping. """ selector = self.parameters.get( "resource_selector", self.parameters.get("selector", {}), ) if selector is None: selector = {} if not isinstance(selector, Mapping): raise TypeError("resource_selector must be a mapping.") normalized = deepcopy(dict(selector)) self._validate_string_keys(normalized, "resource_selector") for parameter_key, selector_key in ( ("resource_id", "id"), ("source_resource_id", "id"), ("resource_type", "type"), ("source_resource_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 _target_selector(self) -> dict[str, Any]: """ Build an agent source selector from parameters. Returns ------- dict[str, Any] Selector mapping. """ selector = self.parameters.get( "target_selector", self.parameters.get( "source_selector", self.parameters.get("agent_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"), ("source_agent_id", "id"), ("mentor_id", "id"), ("target_type", "type"), ("source_agent_type", "type"), ("mentor_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 _amount( self, *, default: float, aliases: Sequence[str] = ( "amount", "quantity", "intensity", "research_amount", ), ) -> float: """ Return a configured non-negative amount. Parameters ---------- default: Default amount. aliases: Parameter keys to inspect. Returns ------- float Non-negative amount. """ for key in aliases: 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 _source_kind(self, *, default: ResearchSourceKind) -> ResearchSourceKind: """ Return normalized research source kind. Parameters ---------- default: Default source kind. Returns ------- ResearchSourceKind ``"self"``, ``"resource"``, or ``"agent"``. """ raw_value = self.parameters.get( "source", self.parameters.get("source_kind", default), ) value = str(raw_value).strip().lower() if value in {"self", "internal", "independent", "solo"}: return "self" if value in {"resource", "resources"}: return "resource" if value in {"agent", "agents", "mentor", "peer"}: return "agent" raise ValueError("source must be one of: self, resource, agent.") def _radius(self) -> float | None: """ Return optional interaction radius. Returns ------- float | None Non-negative radius or None. """ return self._optional_non_negative_float_parameter( "radius", aliases=("range", "interaction_radius", "learning_radius"), default=None, ) def _selection_strategy(self, *, default: SelectionStrategy) -> SelectionStrategy: """ Return selection strategy. Parameters ---------- default: Default strategy. Returns ------- SelectionStrategy Normalized strategy. """ raw_value = self.parameters.get( "selection_strategy", self.parameters.get("target_strategy", default), ) 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 {"largest", "highest_amount", "max_amount"}: return "largest" if value in {"smallest", "lowest_amount", "min_amount"}: return "smallest" 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, " "largest, smallest, highest_state, lowest_state." ) def _apply_numeric_increment( self, *, agent: Agent, pending_updates: dict[str, Any], key: str, amount: float, min_key: str = "min_state_value", max_key: str = "max_state_value", ) -> None: """ Increment a numeric state key in pending updates. Parameters ---------- agent: Agent whose state should be updated. pending_updates: Pending updates mapping. key: State key. amount: Amount to add. min_key: Optional parameter key for lower clamp. max_key: Optional parameter key for upper clamp. """ current = self._numeric_state_with_pending( agent=agent, key=key, pending_updates=pending_updates, default=0.0, ) new_value = current + amount min_value = self._optional_finite_float_parameter( min_key, aliases=(), default=None, ) max_value = self._optional_finite_float_parameter( max_key, aliases=(), 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) pending_updates[key] = new_value def _apply_scaled_output_updates( self, *, agent: Agent, pending_updates: dict[str, Any], actual_amount: float, default_output_key: str | None, ) -> None: """ Apply primary and mapping-based output state updates. Parameters ---------- agent: Acting agent. pending_updates: Pending actor updates. actual_amount: Base amount used to scale outputs. default_output_key: Default primary output key, or None to skip primary output. """ output_key = self._optional_string_parameter( "output_state_key", aliases=("state_gain_key", "gain_state_key", "result_state_key"), default=default_output_key, ) if output_key is not None: multiplier = self._optional_finite_float_parameter( "output_multiplier", aliases=("state_gain_multiplier", "gain_multiplier"), default=1.0, ) assert multiplier is not None self._apply_numeric_increment( agent=agent, pending_updates=pending_updates, key=output_key, amount=actual_amount * multiplier, ) output_state_updates = self.parameters.get("output_state_updates") if output_state_updates is None: output_state_updates = self.parameters.get("state_gain_updates") if output_state_updates is not None: if not isinstance(output_state_updates, Mapping): raise TypeError("output_state_updates must be a mapping.") for key, multiplier_value in output_state_updates.items(): if not isinstance(key, str) or not key.strip(): raise TypeError("output_state_updates keys must be non-empty strings.") multiplier = self._normalize_finite_float( multiplier_value, f"output_state_updates['{key}']", ) self._apply_numeric_increment( agent=agent, pending_updates=pending_updates, key=key.strip(), amount=actual_amount * multiplier, ) 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. Returns ------- bool True if no cost exists or the agent has enough budget. """ 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_agent_state(agent, key, default=0.0) return current >= cost def _build_cost_state_updates( self, *, agent: Agent, intensity: float, pending_updates: Mapping[str, Any], charge: bool = True, ) -> dict[str, Any]: """ Build optional action cost updates. Parameters ---------- agent: Acting agent. intensity: Action intensity. pending_updates: 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 {} current = self._numeric_state_with_pending( agent=agent, key=key, pending_updates=pending_updates, default=0.0, ) return {key: current - cost} def _cost_state_key(self) -> str | None: """ Return optional cost state key. Returns ------- str | None State key charged for research action cost. """ return self._optional_string_parameter( "cost_state_key", aliases=("state_cost_key", "budget_state_key"), default=None, ) def _action_cost(self, intensity: float) -> float: """ Compute optional action cost. Parameters ---------- intensity: Action intensity. Returns ------- float Non-negative 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_intensity"), 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 _asset_list_from_parameters( self, *, primary_keys: Sequence[str], default_container: AssetContainer = "state", ) -> tuple[ResearchAsset, ...]: """ Read a sequence of ResearchAsset objects from behavior parameters. Parameters ---------- primary_keys: Parameter keys to search. default_container: Default asset container. Returns ------- tuple[ResearchAsset, ...] Normalized assets. """ raw_value: Any = None found = False for key in primary_keys: if key in self.parameters and self.parameters[key] is not None: raw_value = self.parameters[key] found = True break if not found: 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( f"Parameter '{primary_keys[0]}' must be a mapping or sequence." ) return tuple( ResearchAsset.from_mapping( raw_asset, default_container=default_container, default_inventory_key=self._inventory_key(), ) for raw_asset in raw_assets ) def _available_asset_fraction( self, *, agent: Agent, assets: Sequence[ResearchAsset], pending_updates: Mapping[str, Any], ) -> float: """ Compute available fraction for a list of assets. Parameters ---------- agent: Agent whose assets should be checked. assets: Requested assets. pending_updates: Pending 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) def _consume_assets( self, *, agent: Agent, pending_updates: dict[str, Any], assets: Sequence[ResearchAsset], ) -> None: """ Subtract assets from an agent into pending updates. Parameters ---------- agent: Agent consuming assets. pending_updates: Pending updates. assets: Assets to consume. """ for asset in assets: current = self._read_asset_amount( agent=agent, asset=asset, pending_updates=pending_updates, ) new_amount = current - asset.amount if new_amount < -ZERO_TOLERANCE: raise ValueError( f"Agent '{agent.id}' lacks asset '{asset.key}'. " f"Required {asset.amount}, available {current}." ) self._write_asset_amount( agent=agent, pending_updates=pending_updates, asset=asset, amount=max(0.0, new_amount), ) def _produce_assets( self, *, agent: Agent, pending_updates: dict[str, Any], assets: Sequence[ResearchAsset], ) -> None: """ Add assets to an agent into pending updates. Parameters ---------- agent: Agent producing assets. pending_updates: Pending updates. assets: Assets to add. """ for asset in assets: 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 + asset.amount, ) def _read_asset_amount( self, *, agent: Agent, asset: ResearchAsset, 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 asset 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: ResearchAsset, amount: float, ) -> None: """ Write asset amount to pending updates. Parameters ---------- agent: Agent whose state is updated. pending_updates: Pending 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 _read_inventory( self, *, agent: Agent, inventory_key: str, pending_updates: Mapping[str, Any], ) -> dict[str, Any]: """ Read inventory mapping from state plus pending updates. Parameters ---------- agent: Agent whose inventory should be read. inventory_key: Inventory state key. pending_updates: Pending 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 _create_resource_outputs( self, *, agent: Agent, world: WorldProtocol, actual_amount: float, output_kind: str, ) -> tuple[dict[str, Any], ...]: """ Create optional world resources from research output specifications. Parameters ---------- agent: Acting agent. world: World receiving resources. actual_amount: Base amount used by output specs. output_kind: Research behavior kind. Returns ------- tuple[dict[str, Any], ...] Created resource snapshots. """ specs = self._resource_output_specs() if not specs: return tuple() add_resource = getattr(world, "add_resource", None) if not callable(add_resource): raise TypeError("World must implement add_resource(...) for resource outputs.") created: list[dict[str, Any]] = [] for index, spec in enumerate(specs): resource_type = ( spec.get("type") or spec.get("resource_type") or spec.get("output_resource_type") ) if resource_type is None: raise KeyError("Resource output spec requires 'type' or 'resource_type'.") amount = spec.get("amount") if amount is None: multiplier = self._normalize_finite_float( spec.get("amount_multiplier", 1.0), "amount_multiplier", ) amount = actual_amount * multiplier normalized_amount = self._normalize_non_negative_float(amount, "amount") position = spec.get("position", agent.position.copy()) metadata = spec.get("metadata", {}) if metadata is None: metadata = {} if not isinstance(metadata, Mapping): raise TypeError("Resource output metadata must be a mapping.") resource_id = spec.get("id") if resource_id is None: resource_id = self._unique_resource_id( world=world, agent=agent, resource_type=str(resource_type), output_kind=output_kind, index=index, ) resource = Resource( id=str(resource_id), type=str(resource_type), amount=normalized_amount, position=position, metadata={ **deepcopy(dict(metadata)), "created_by_behavior": self.id, "created_by_agent": agent.id, "research_kind": output_kind, }, ) add_resource(resource) created.append(resource.snapshot()) return tuple(created) def _resource_output_specs(self) -> tuple[Mapping[str, Any], ...]: """ Return configured world-resource output specs. Returns ------- tuple[Mapping[str, Any], ...] Output resource specs. """ raw_value: Any = None found = False for key in ( "resource_outputs", "resource_output", "created_resources", "created_resource", ): if key in self.parameters and self.parameters[key] is not None: raw_value = self.parameters[key] found = True break if not found: if self._optional_bool_parameter("create_resource", default=False): resource_type = self._required_string_parameter( "output_resource_type", aliases=("resource_type",), ) return ( { "type": resource_type, "amount_multiplier": self.parameters.get( "resource_amount_multiplier", 1.0, ), "metadata": self.parameters.get("resource_metadata", {}), }, ) return tuple() if isinstance(raw_value, Mapping): return (deepcopy(dict(raw_value)),) if isinstance(raw_value, Sequence) and not isinstance(raw_value, (str, bytes)): specs: list[Mapping[str, Any]] = [] for item in raw_value: if not isinstance(item, Mapping): raise TypeError("Resource output specs must be mappings.") specs.append(deepcopy(dict(item))) return tuple(specs) raise TypeError("resource_outputs must be a mapping or sequence of mappings.") def _unique_resource_id( self, *, world: WorldProtocol, agent: Agent, resource_type: str, output_kind: str, index: int, ) -> str: """ Build a collision-resistant resource id. Parameters ---------- world: World containing existing resources. agent: Acting agent. resource_type: Resource type. output_kind: Research behavior kind. index: Output index. Returns ------- str Resource id. """ step_count = getattr(world, "step_count", 0) base = f"{agent.id}_{self.id}_{output_kind}_{resource_type}_{step_count}_{index}" normalized = "_".join(base.lower().split()) resources = getattr(world, "resources", {}) if not isinstance(resources, Mapping) or normalized not in resources: return normalized suffix = 1 while f"{normalized}_{suffix}" in resources: suffix += 1 return f"{normalized}_{suffix}" def _publication_like_execution( self, *, agent: Agent, world: WorldProtocol, research_kind: str, default_output_key: str, success_message: str, failure_message: str, ) -> BehaviorResult: """ Execute a publish/patent style behavior. Parameters ---------- agent: Acting agent. world: World in which output may be created. research_kind: Behavior kind. default_output_key: Default state key incremented by output. success_message: Success message. failure_message: Failure message. Returns ------- BehaviorResult Behavior execution result. """ try: requested_amount = self._amount( default=DEFAULT_RESEARCH_AMOUNT, aliases=("amount", "input_amount", "knowledge_amount"), ) input_key = self._optional_string_parameter( "input_state_key", aliases=("source_state_key", "knowledge_state_key"), default=DEFAULT_KNOWLEDGE_STATE_KEY, ) assert input_key is not None available = self._numeric_agent_state(agent, input_key, default=0.0) allow_partial = self._optional_bool_parameter("allow_partial", default=False) if requested_amount <= ZERO_TOLERANCE: fraction = 1.0 else: fraction = min(1.0, available / requested_amount) if not allow_partial and fraction < 1.0 - ZERO_TOLERANCE: return self.failure( reward=self._failure_reward(), message=failure_message, metadata={ "reason": "insufficient_input_state", "input_state_key": input_key, "required": requested_amount, "available": available, }, ) minimum_fraction = self._minimum_fraction( key="minimum_output_fraction", default=ZERO_TOLERANCE, ) if allow_partial and fraction < minimum_fraction: return self.failure( reward=self._failure_reward(), message=failure_message, metadata={ "reason": "below_minimum_fraction", "fraction": fraction, "minimum_fraction": minimum_fraction, }, ) actual_amount = requested_amount * max(0.0, min(1.0, fraction)) if actual_amount <= ZERO_TOLERANCE and requested_amount > ZERO_TOLERANCE: return self.failure( reward=self._failure_reward(), message=failure_message, metadata={"reason": "zero_actual_amount"}, ) if not self._has_sufficient_action_budget( agent=agent, intensity=actual_amount, ): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget.", metadata={ "required_cost": self._action_cost(actual_amount), "cost_state_key": self._cost_state_key(), }, ) actor_updates: dict[str, Any] = {} if self._optional_bool_parameter("consume_input", default=True): actor_updates[input_key] = max(0.0, available - actual_amount) self._apply_scaled_output_updates( agent=agent, pending_updates=actor_updates, actual_amount=actual_amount, default_output_key=default_output_key, ) actor_updates.update( self._build_cost_state_updates( agent=agent, intensity=actual_amount, pending_updates=actor_updates, charge=True, ) ) created_resources = self._create_resource_outputs( agent=agent, world=world, actual_amount=actual_amount, output_kind=research_kind, ) computation = ResearchComputation( research_kind=research_kind, actor_id=agent.id, actor_type=agent.type, requested_amount=requested_amount, actual_amount=actual_amount, actor_state_updates=actor_updates, created_resources=created_resources, partial=fraction < 1.0 - ZERO_TOLERANCE, metadata={ "input_state_key": input_key, "output_state_key_default": default_output_key, "consume_input": self._optional_bool_parameter( "consume_input", default=True, ), "fraction": fraction, }, ) metadata = computation.to_metadata() return self.success( reward=self._success_reward(intensity=actual_amount, fraction=fraction), state_updates=actor_updates, memory_updates={"research": metadata}, message=success_message, metadata=metadata, ) except Exception as exc: logger.exception( "%s failed for agent_id=%s behavior_id=%s", self.__class__.__name__, agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message=f"{self.__class__.__name__} raised an exception.", metadata={"error": repr(exc), "research_kind": research_kind}, ) def _minimum_fraction(self, *, key: str, default: float) -> float: """ Return a minimum fraction parameter. Parameters ---------- key: Parameter key. default: Default value. Returns ------- float Fraction in [0, 1]. """ value = self._optional_non_negative_float_parameter( key, aliases=("minimum_fraction", "min_fraction"), default=default, ) assert value is not None if value > 1.0: raise ValueError(f"{key} cannot exceed 1.0.") return value def _success_reward(self, *, intensity: float = 0.0, fraction: float = 1.0) -> float: """ Compute success reward. Parameters ---------- intensity: Behavior intensity. fraction: Completed fraction. Returns ------- float 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_amount", "reward_per_intensity"), 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 = self._optional_finite_float_parameter( "failure_reward", aliases=("failure_penalty",), default=0.0, ) assert value is not None return value def _inventory_key(self) -> str: """ Return default inventory key. Returns ------- str Inventory state key. """ value = self._optional_string_parameter( "inventory_key", aliases=("storage_key",), default=DEFAULT_INVENTORY_KEY, ) assert value is not None return value @staticmethod def _numeric_agent_state(agent: Agent, key: str, *, default: float) -> float: """ Read a numeric state value from an agent. Parameters ---------- agent: Agent whose state should be read. key: State key. default: Default value if missing. Returns ------- float Numeric state value. """ value = agent.state.get(key, 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 def _numeric_state_with_pending( self, *, agent: Agent, key: str, pending_updates: Mapping[str, Any], default: float, ) -> float: """ Read numeric state from pending updates or agent state. Parameters ---------- agent: Agent whose state should be read. key: State key. pending_updates: Pending updates. default: Default value. Returns ------- float Numeric value. """ value = pending_updates.get(key, agent.state.get(key, default)) if isinstance(value, bool) or not isinstance(value, Real): raise TypeError(f"State key '{key}' must be numeric.") normalized = float(value) if not np.isfinite(normalized): raise ValueError(f"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 numeric thresholds are satisfied. Parameters ---------- mapping: Candidate mapping. thresholds: Threshold mapping. comparison: ``"min"`` or ``"max"``. Returns ------- bool True if all 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 thresholds 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 _roll_success(self, world: WorldProtocol) -> bool: """ Resolve optional stochastic success. Parameters ---------- world: World providing optional RNG. Returns ------- bool True if the behavior succeeds. """ probability = self._optional_non_negative_float_parameter( "success_probability", aliases=("probability", "innovation_probability"), default=1.0, ) assert probability is not None if probability > 1.0: raise ValueError("success_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 _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 StudyBehavior(ResearchBehavior): """ Study independently or from a world resource. Study increases a configurable actor state key, usually representing knowledge, skill, mastery, insight, or any DSL-defined learning quantity. Common parameters ----------------- source: ``"self"`` or ``"resource"``. Defaults to ``"resource"``. resource_selector: Selector used when source is ``"resource"``. amount: Study amount. deplete_resource: Whether studying reduces the selected resource. Defaults to False. state_gain_key: Actor state key to increase. Defaults to ``"knowledge"``. state_gain_multiplier: Multiplier applied to actual studied amount. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check study-specific preconditions. Parameters ---------- agent: Acting agent. world: World containing optional study resources. Returns ------- bool True if study can be attempted. """ try: amount = self._amount(default=DEFAULT_RESEARCH_AMOUNT) if not self._has_sufficient_action_budget(agent=agent, intensity=amount): return False source_kind = self._source_kind(default="resource") if source_kind == "self": return True if source_kind == "resource": if self._optional_bool_parameter("require_source", default=True): return self._select_resource(agent=agent, world=world) is not None return True return False except Exception: logger.exception( "Study precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute study behavior. Parameters ---------- agent: Acting agent. world: World containing optional study resources. Returns ------- BehaviorResult Study result. """ try: requested_amount = self._amount(default=DEFAULT_RESEARCH_AMOUNT) source_kind = self._source_kind(default="resource") actual_amount = requested_amount partial = False source: ResearchCandidate | None = None resource_results: list[ResourceOperationResult] = [] if source_kind == "resource": selected = self._select_resource(agent=agent, world=world) if selected is None: if self._optional_bool_parameter("require_source", default=True): return self.failure( reward=self._failure_reward(), message="StudyBehavior found no matching resource source.", metadata={ "resource_selector": self._resource_selector(), "radius": self._radius(), }, ) else: resource, source = selected deplete_resource = self._optional_bool_parameter( "deplete_resource", default=False, ) allow_partial = self._optional_bool_parameter( "allow_partial", default=True, ) if deplete_resource: operation_result = resource.deplete( requested_amount, allow_partial=allow_partial, metadata={ "source": self.__class__.__name__, "behavior_id": self.id, "agent_id": agent.id, }, ) resource_results.append(operation_result) actual_amount = abs(operation_result.actual_delta) partial = operation_result.partial else: if self._optional_bool_parameter( "require_available_amount", default=True, ): actual_amount = min(requested_amount, resource.amount) partial = actual_amount < requested_amount else: actual_amount = requested_amount partial = False if partial and not allow_partial: return self.failure( reward=self._failure_reward(), message="StudyBehavior could not complete full study amount.", metadata={ "requested_amount": requested_amount, "actual_amount": actual_amount, "source": source.to_metadata() if source else None, }, ) if actual_amount <= ZERO_TOLERANCE and requested_amount > ZERO_TOLERANCE: return self.failure( reward=self._failure_reward(), message="StudyBehavior produced zero study amount.", metadata={"requested_amount": requested_amount}, ) if not self._has_sufficient_action_budget( agent=agent, intensity=actual_amount, ): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget to study.", metadata={ "required_cost": self._action_cost(actual_amount), "cost_state_key": self._cost_state_key(), }, ) actor_updates: dict[str, Any] = {} self._apply_scaled_output_updates( agent=agent, pending_updates=actor_updates, actual_amount=actual_amount, default_output_key=DEFAULT_KNOWLEDGE_STATE_KEY, ) actor_updates.update( self._build_cost_state_updates( agent=agent, intensity=actual_amount, pending_updates=actor_updates, charge=True, ) ) created_resources = self._create_resource_outputs( agent=agent, world=world, actual_amount=actual_amount, output_kind="study", ) computation = ResearchComputation( research_kind="study", actor_id=agent.id, actor_type=agent.type, requested_amount=requested_amount, actual_amount=actual_amount, source=source, actor_state_updates=actor_updates, resource_results=tuple(resource_results), created_resources=created_resources, partial=partial, metadata={ "source_kind": source_kind, "deplete_resource": self._optional_bool_parameter( "deplete_resource", default=False, ), }, ) metadata = computation.to_metadata() return self.success( reward=self._success_reward(intensity=actual_amount), state_updates=actor_updates, memory_updates={"research": metadata}, message="Agent studied.", metadata=metadata, ) except Exception as exc: logger.exception( "StudyBehavior failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message="StudyBehavior raised an exception.", metadata={"error": repr(exc)}, ) @dataclass(slots=True) class LearnBehavior(ResearchBehavior): """ Learn from another agent. LearnBehavior transfers or copies a configurable state quantity from a source agent to the acting agent. By default, learning does not reduce the source agent's state. Common parameters ----------------- target_selector: Selector for source agent. source_state_key: State key read from source agent. Defaults to ``"knowledge"``. state_gain_key: Actor state key to increase. Defaults to ``"knowledge"``. amount: Learning amount. decrement_source: Whether learning reduces the source state. Defaults to False. target_reward_state_key: Optional state key incremented on the source agent. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check learn-specific preconditions. Parameters ---------- agent: Acting agent. world: World containing source agents. Returns ------- bool True if a source agent and action budget are available. """ try: amount = self._amount(default=DEFAULT_RESEARCH_AMOUNT) if not self._has_sufficient_action_budget(agent=agent, intensity=amount): return False if self._optional_bool_parameter("require_target", default=True): return self._select_agent_source(agent=agent, world=world) is not None return True except Exception: logger.exception( "Learn precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute learning behavior. Parameters ---------- agent: Acting learner. world: World containing possible source agents. Returns ------- BehaviorResult Learn result. """ try: requested_amount = self._amount(default=DEFAULT_RESEARCH_AMOUNT) selected = self._select_agent_source(agent=agent, world=world) if selected is None: if self._optional_bool_parameter("require_target", default=True): return self.failure( reward=self._failure_reward(), message="LearnBehavior found no matching source agent.", metadata={ "target_selector": self._target_selector(), "radius": self._radius(), }, ) source_agent = None source = None actual_amount = requested_amount partial = False else: source_agent, source = selected source_state_key = self._optional_string_parameter( "source_state_key", aliases=("knowledge_state_key", "target_state_key"), default=DEFAULT_KNOWLEDGE_STATE_KEY, ) assert source_state_key is not None available = self._numeric_agent_state( source_agent, source_state_key, default=0.0, ) if self._optional_bool_parameter("require_available_amount", default=True): actual_amount = min(requested_amount, available) else: actual_amount = requested_amount partial = actual_amount < requested_amount allow_partial = self._optional_bool_parameter("allow_partial", default=True) if partial and not allow_partial: return self.failure( reward=self._failure_reward(), message="LearnBehavior source lacks required amount.", metadata={ "source_state_key": source_state_key, "requested_amount": requested_amount, "available": available, }, ) if actual_amount <= ZERO_TOLERANCE and requested_amount > ZERO_TOLERANCE: return self.failure( reward=self._failure_reward(), message="LearnBehavior produced zero learning amount.", metadata={"requested_amount": requested_amount}, ) if not self._has_sufficient_action_budget( agent=agent, intensity=actual_amount, ): return self.failure( reward=self._failure_reward(), message="Agent does not have sufficient state budget to learn.", metadata={ "required_cost": self._action_cost(actual_amount), "cost_state_key": self._cost_state_key(), }, ) actor_updates: dict[str, Any] = {} target_updates: dict[str, Any] = {} self._apply_scaled_output_updates( agent=agent, pending_updates=actor_updates, actual_amount=actual_amount, default_output_key=DEFAULT_KNOWLEDGE_STATE_KEY, ) if source_agent is not None: source_state_key = self._optional_string_parameter( "source_state_key", aliases=("knowledge_state_key", "target_state_key"), default=DEFAULT_KNOWLEDGE_STATE_KEY, ) assert source_state_key is not None if self._optional_bool_parameter("decrement_source", default=False): current_source_value = self._numeric_agent_state( source_agent, source_state_key, default=0.0, ) target_updates[source_state_key] = max( 0.0, current_source_value - actual_amount, ) reward_key = self._optional_string_parameter( "target_reward_state_key", aliases=("teacher_reward_state_key", "mentor_reward_state_key"), default=None, ) if reward_key is not None: reward_multiplier = self._optional_finite_float_parameter( "target_reward_multiplier", aliases=("teacher_reward_multiplier",), default=1.0, ) assert reward_multiplier is not None current_reward_value = self._numeric_state_with_pending( agent=source_agent, key=reward_key, pending_updates=target_updates, default=0.0, ) target_updates[reward_key] = current_reward_value + ( actual_amount * reward_multiplier ) if target_updates: source_agent.update_state(target_updates) actor_updates.update( self._build_cost_state_updates( agent=agent, intensity=actual_amount, pending_updates=actor_updates, charge=True, ) ) created_resources = self._create_resource_outputs( agent=agent, world=world, actual_amount=actual_amount, output_kind="learn", ) computation = ResearchComputation( research_kind="learn", actor_id=agent.id, actor_type=agent.type, requested_amount=requested_amount, actual_amount=actual_amount, source=source, actor_state_updates=actor_updates, target_state_updates=target_updates, created_resources=created_resources, partial=partial, metadata={ "decrement_source": self._optional_bool_parameter( "decrement_source", default=False, ), }, ) metadata = computation.to_metadata() return self.success( reward=self._success_reward(intensity=actual_amount), state_updates=actor_updates, memory_updates={"research": metadata}, message="Agent learned from source.", metadata=metadata, ) except Exception as exc: logger.exception( "LearnBehavior failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message="LearnBehavior raised an exception.", metadata={"error": repr(exc)}, ) @dataclass(slots=True) class PublishBehavior(ResearchBehavior): """ Publish research output. PublishBehavior converts internal actor state into configurable output state, such as reputation, influence, credibility, citations, public knowledge, or any DSL-defined quantity. Common parameters ----------------- input_state_key: State key required for publication. Defaults to ``"knowledge"``. amount: Required publication input amount. consume_input: Whether input state is reduced. Defaults to True. output_state_key: State key incremented by publication. Defaults to ``"reputation"``. output_multiplier: Multiplier applied to actual published amount. resource_outputs: Optional world resources created by publication. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute publication behavior. Parameters ---------- agent: Acting agent. world: World receiving optional public resources. Returns ------- BehaviorResult Publication result. """ return self._publication_like_execution( agent=agent, world=world, research_kind="publish", default_output_key=DEFAULT_REPUTATION_STATE_KEY, success_message="Agent published output.", failure_message="PublishBehavior could not publish output.", ) @dataclass(slots=True) class PatentBehavior(ResearchBehavior): """ Patent, claim, or protect an invention-like output. PatentBehavior is generic. It does not model any real legal system. It simply converts input state into a configurable protected-output state such as patents, claims, intellectual property, protected discoveries, exclusive recipes, or invention rights. Common parameters ----------------- input_state_key: State key required for patenting. Defaults to ``"knowledge"``. amount: Required input amount. consume_input: Whether input state is reduced. Defaults to True. output_state_key: State key incremented by patenting. Defaults to ``"patents"``. output_multiplier: Multiplier applied to actual patented amount. output_state_updates: Optional additional output state multipliers. """ def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute patent behavior. Parameters ---------- agent: Acting agent. world: World receiving optional patent resources. Returns ------- BehaviorResult Patent result. """ return self._publication_like_execution( agent=agent, world=world, research_kind="patent", default_output_key=DEFAULT_PATENT_STATE_KEY, success_message="Agent patented output.", failure_message="PatentBehavior could not patent output.", ) @dataclass(slots=True) class InnovateBehavior(ResearchBehavior): """ Transform inputs into new outputs. InnovateBehavior is a generic recipe-like transformation. It can consume state assets, inventory assets, or no assets, then produce state assets, inventory assets, and optional world resources. Common parameters ----------------- inputs: Assets consumed from the actor. outputs: Assets produced for the actor. allow_partial: Whether partial innovation is allowed when inputs are insufficient. Defaults to False. consume_inputs: Whether configured inputs are consumed. Defaults to True. success_probability: Probability of innovation success. Defaults to 1.0. output_state_key: Default state output if no explicit outputs are configured. Defaults to ``"innovation"``. """ def _check_preconditions(self, agent: Agent, world: WorldProtocol) -> bool: """ Check innovation-specific preconditions. Parameters ---------- agent: Acting agent. world: World in which innovation occurs. Returns ------- bool True if inputs and budget allow innovation attempt. """ try: inputs = self._innovation_inputs() fraction = self._input_fraction(agent=agent, inputs=inputs) if inputs and fraction <= ZERO_TOLERANCE: return False intensity = self._innovation_intensity(inputs=inputs, fraction=fraction) return self._has_sufficient_action_budget(agent=agent, intensity=intensity) except Exception: logger.exception( "Innovate precondition failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return False def execute(self, agent: Agent, world: WorldProtocol) -> BehaviorResult: """ Execute innovation behavior. Parameters ---------- agent: Acting agent. world: World receiving optional innovation resources. Returns ------- BehaviorResult Innovation result. """ try: requested_inputs = self._innovation_inputs() fraction = self._input_fraction(agent=agent, inputs=requested_inputs) allow_partial = self._optional_bool_parameter("allow_partial", default=False) if requested_inputs and not allow_partial and fraction < 1.0 - ZERO_TOLERANCE: return self.failure( reward=self._failure_reward(), message="InnovateBehavior lacks required inputs.", metadata={ "requested_inputs": [ asset.to_metadata() for asset in requested_inputs ], "available_fraction": fraction, }, ) if allow_partial: minimum_fraction = self._minimum_fraction( key="minimum_innovation_fraction", default=ZERO_TOLERANCE, ) if fraction < minimum_fraction: return self.failure( reward=self._failure_reward(), message="InnovateBehavior input fraction is below minimum.", metadata={ "available_fraction": fraction, "minimum_fraction": minimum_fraction, }, ) executed_fraction = max(0.0, min(1.0, fraction)) consumed_inputs = tuple(asset.scaled(executed_fraction) for asset in requested_inputs) intensity = self._innovation_intensity( inputs=requested_inputs, fraction=executed_fraction, ) 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 innovate.", metadata={ "required_cost": self._action_cost(intensity), "cost_state_key": self._cost_state_key(), }, ) actor_updates: dict[str, Any] = {} if self._optional_bool_parameter("consume_inputs", default=True): self._consume_assets( agent=agent, pending_updates=actor_updates, assets=consumed_inputs, ) actor_updates.update( self._build_cost_state_updates( agent=agent, intensity=intensity, pending_updates=actor_updates, charge=True, ) ) if not self._roll_success(world): computation = ResearchComputation( research_kind="innovate", actor_id=agent.id, actor_type=agent.type, requested_amount=intensity, actual_amount=0.0, actor_state_updates=actor_updates, consumed_assets=consumed_inputs, partial=executed_fraction < 1.0 - ZERO_TOLERANCE, metadata={ "success_probability": self.parameters.get( "success_probability", 1.0, ), "succeeded": False, }, ) metadata = computation.to_metadata() return self.failure( reward=self._failure_reward(), state_updates=actor_updates, memory_updates={"research": metadata}, message="InnovateBehavior attempt did not succeed.", metadata=metadata, ) produced_outputs = self._innovation_outputs( actual_input_total=intensity, fraction=executed_fraction, ) self._produce_assets( agent=agent, pending_updates=actor_updates, assets=produced_outputs, ) self._apply_scaled_output_updates( agent=agent, pending_updates=actor_updates, actual_amount=intensity, default_output_key=None, ) created_resources = self._create_resource_outputs( agent=agent, world=world, actual_amount=intensity, output_kind="innovate", ) computation = ResearchComputation( research_kind="innovate", actor_id=agent.id, actor_type=agent.type, requested_amount=sum(asset.amount for asset in requested_inputs), actual_amount=intensity, actor_state_updates=actor_updates, consumed_assets=consumed_inputs, produced_assets=produced_outputs, created_resources=created_resources, partial=executed_fraction < 1.0 - ZERO_TOLERANCE, metadata={ "input_fraction": executed_fraction, "consume_inputs": self._optional_bool_parameter( "consume_inputs", default=True, ), "succeeded": True, }, ) metadata = computation.to_metadata() return self.success( reward=self._success_reward( intensity=intensity, fraction=executed_fraction, ), state_updates=actor_updates, memory_updates={"research": metadata}, message="Agent innovated.", metadata=metadata, ) except Exception as exc: logger.exception( "InnovateBehavior failed for agent_id=%s behavior_id=%s", agent.id, self.id, ) return self.failure( reward=self._failure_reward(), message="InnovateBehavior raised an exception.", metadata={"error": repr(exc)}, ) def _innovation_inputs(self) -> tuple[ResearchAsset, ...]: """ Return configured innovation input assets. Returns ------- tuple[ResearchAsset, ...] Input assets. """ explicit_inputs = self._asset_list_from_parameters( primary_keys=("inputs", "input_assets", "required_assets"), default_container="state", ) if explicit_inputs: return explicit_inputs if "input_state_key" in self.parameters or "amount" in self.parameters: input_key = self._optional_string_parameter( "input_state_key", aliases=("source_state_key", "knowledge_state_key"), default=DEFAULT_KNOWLEDGE_STATE_KEY, ) assert input_key is not None amount = self._amount( default=DEFAULT_RESEARCH_AMOUNT, aliases=("amount", "input_amount", "knowledge_amount"), ) return ( ResearchAsset( key=input_key, amount=amount, container="state", inventory_key=self._inventory_key(), metadata={"role": "default_input"}, ), ) return tuple() def _innovation_outputs( self, *, actual_input_total: float, fraction: float, ) -> tuple[ResearchAsset, ...]: """ Return configured innovation output assets. Parameters ---------- actual_input_total: Actual innovation intensity. fraction: Executed input fraction. Returns ------- tuple[ResearchAsset, ...] Output assets. """ explicit_outputs = self._asset_list_from_parameters( primary_keys=("outputs", "output_assets", "produced_assets"), default_container="state", ) if explicit_outputs: if self._optional_bool_parameter("scale_outputs_by_fraction", default=True): return tuple(asset.scaled(fraction) for asset in explicit_outputs) return explicit_outputs output_key = self._optional_string_parameter( "output_state_key", aliases=("innovation_state_key", "result_state_key"), default=DEFAULT_INNOVATION_STATE_KEY, ) assert output_key is not None output_amount = self._optional_non_negative_float_parameter( "output_amount", aliases=("innovation_amount", "result_amount"), default=None, ) if output_amount is None: output_multiplier = self._optional_finite_float_parameter( "output_multiplier", aliases=("innovation_multiplier",), default=1.0, ) assert output_multiplier is not None output_amount = max(DEFAULT_RESEARCH_AMOUNT, actual_input_total * output_multiplier) return ( ResearchAsset( key=output_key, amount=output_amount * fraction, container="state", inventory_key=self._inventory_key(), metadata={"role": "default_output"}, ), ) def _input_fraction( self, *, agent: Agent, inputs: Sequence[ResearchAsset], ) -> float: """ Compute available input fraction. Parameters ---------- agent: Acting agent. inputs: Requested inputs. Returns ------- float Available fraction. """ if not inputs: return 1.0 return min( 1.0, self._available_asset_fraction( agent=agent, assets=inputs, pending_updates={}, ), ) @staticmethod def _innovation_intensity( *, inputs: Sequence[ResearchAsset], fraction: float, ) -> float: """ Compute innovation intensity from inputs and execution fraction. Parameters ---------- inputs: Requested input assets. fraction: Executed fraction. Returns ------- float Non-negative intensity. """ total = sum(asset.amount for asset in inputs) if total <= ZERO_TOLERANCE: return DEFAULT_RESEARCH_AMOUNT * fraction return total * fraction Study = StudyBehavior Learn = LearnBehavior Publish = PublishBehavior Patent = PatentBehavior Innovate = InnovateBehavior __all__ = [ "AssetContainer", "DEFAULT_INNOVATION_STATE_KEY", "DEFAULT_INVENTORY_KEY", "DEFAULT_KNOWLEDGE_STATE_KEY", "DEFAULT_PATENT_STATE_KEY", "DEFAULT_REPUTATION_STATE_KEY", "DEFAULT_RESEARCH_AMOUNT", "Innovate", "InnovateBehavior", "Learn", "LearnBehavior", "Patent", "PatentBehavior", "Publish", "PublishBehavior", "ResearchAsset", "ResearchBehavior", "ResearchCandidate", "ResearchComputation", "ResearchSourceKind", "SelectionStrategy", "Study", "StudyBehavior", ]