""" behaviors.collaboration ================ Generic collaboration behaviors for WorldSmithAI. This module contains domain-agnostic behaviors that let agents cooperate, communicate, share state or memory, recommend options, and negotiate structured agreements. The behaviors intentionally avoid assumptions about species, professions, economies, governments, or social roles. Example: behavior = CommunicateBehavior( message="status_update", content_state_keys=("energy", "health"), ) result = behavior.execute(agent, world) Future extensibility: - Add reputation-aware partner selection without changing the World engine. - Add semantic message payloads produced by an SLM while keeping execution deterministic. - Add negotiation protocols such as auctions, voting, mediation, and contracts. - Connect outcomes to event streams, metrics, and narrative summaries. """ from __future__ import annotations import copy import logging from collections.abc import Iterable, Mapping, MutableMapping, MutableSequence, Sequence from dataclasses import dataclass, field from enum import Enum from numbers import Real from types import MappingProxyType from typing import Any, ClassVar, TYPE_CHECKING import numpy as np from core.behavior import Behavior if TYPE_CHECKING: from core.agent import Agent from core.world import World logger = logging.getLogger(__name__) class ShareLocation(str, Enum): """Supported places from which a value can be shared or into which it can be written.""" STATE = "state" MEMORY = "memory" @dataclass(frozen=True) class BehaviorOutcome: """Serializable execution result emitted by collaboration behaviors. The simulation engine may ignore this object, but returning structured outcomes makes the behavior useful for metrics, debugging, event logs, visualizations, and narrative summaries. """ behavior: str actor_id: str success: bool target_ids: tuple[str, ...] = () step: int | None = None details: Mapping[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly dictionary representation of the outcome.""" return { "behavior": self.behavior, "actor_id": self.actor_id, "success": self.success, "target_ids": list(self.target_ids), "step": self.step, "details": copy.deepcopy(dict(self.details)), } def _agent_id(agent: Agent) -> str: """Return the stable string identifier for an agent.""" return str(getattr(agent, "id")) def _agent_type(agent: Agent) -> str: """Return the generic type label for an agent.""" return str(getattr(agent, "type", "")) def _is_alive(agent: Agent) -> bool: """Return whether an agent should participate in behavior execution.""" return bool(getattr(agent, "alive", True)) def _world_step(world: World) -> int | None: """Return the current simulation step if the world exposes one.""" value = getattr(world, "step_count", None) return int(value) if isinstance(value, Real) and not isinstance(value, bool) else None def _is_number(value: Any) -> bool: """Return whether a value is a real numeric scalar, excluding booleans.""" return isinstance(value, (Real, np.integer, np.floating)) and not isinstance(value, bool) def _as_float(value: Any, default: float = 0.0) -> float: """Convert numeric-like values to float while using a safe default.""" if _is_number(value): return float(value) return default def _as_int(value: Any, default: int = 0) -> int: """Convert numeric-like values to int while using a safe default.""" if _is_number(value): return int(value) return default def _normalize_location(location: ShareLocation | str) -> ShareLocation: """Normalize a state-or-memory location value.""" if isinstance(location, ShareLocation): return location return ShareLocation(location) def _agent_state(agent: Agent) -> MutableMapping[str, Any]: """Return the mutable state mapping for an agent, creating one if needed.""" state = getattr(agent, "state", None) if isinstance(state, MutableMapping): return state replacement: dict[str, Any] = {} setattr(agent, "state", replacement) return replacement def _agent_memory(agent: Agent) -> MutableMapping[str, Any]: """Return the mutable memory mapping for an agent, creating one if needed.""" memory = getattr(agent, "memory", None) if isinstance(memory, MutableMapping): return memory replacement: dict[str, Any] = {} setattr(agent, "memory", replacement) return replacement def _ensure_mapping(parent: MutableMapping[str, Any], key: str) -> MutableMapping[str, Any]: """Return a nested mutable mapping under ``key``, creating one if absent.""" value = parent.get(key) if isinstance(value, MutableMapping): return value replacement: dict[str, Any] = {} parent[key] = replacement return replacement def _ensure_list(parent: MutableMapping[str, Any], key: str) -> MutableSequence[Any]: """Return a nested mutable sequence under ``key``, creating one if absent.""" value = parent.get(key) if isinstance(value, MutableSequence): return value replacement: list[Any] = [] parent[key] = replacement return replacement def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None: """Append an item and trim old records when a positive max size is configured.""" items.append(value) if max_items > 0 and len(items) > max_items: del items[: len(items) - max_items] def _add_unique(items: MutableSequence[Any], value: Any) -> None: """Append a value only if it is not already present.""" if value not in items: items.append(value) def _split_path(path: str) -> tuple[str, ...]: """Split a dot-separated state or memory key into path components.""" return tuple(part for part in path.split(".") if part) def _get_path(container: Mapping[str, Any], path: str, default: Any = None) -> Any: """Read a possibly nested value from a mapping using dot notation.""" parts = _split_path(path) if not parts: return default current: Any = container for part in parts: if not isinstance(current, Mapping) or part not in current: return default current = current[part] return current def _set_path(container: MutableMapping[str, Any], path: str, value: Any) -> None: """Write a possibly nested value to a mapping using dot notation.""" parts = _split_path(path) if not parts: return current: MutableMapping[str, Any] = container for part in parts[:-1]: nested = current.get(part) if not isinstance(nested, MutableMapping): nested = {} current[part] = nested current = nested current[parts[-1]] = value def _delete_path(container: MutableMapping[str, Any], path: str) -> None: """Delete a possibly nested value from a mapping using dot notation.""" parts = _split_path(path) if not parts: return current: MutableMapping[str, Any] = container for part in parts[:-1]: nested = current.get(part) if not isinstance(nested, MutableMapping): return current = nested current.pop(parts[-1], None) def _increment_path(container: MutableMapping[str, Any], path: str, delta: float) -> float: """Increment a numeric value at a nested path and return the updated value.""" current_value = _get_path(container, path, 0.0) updated_value = _as_float(current_value) + delta _set_path(container, path, updated_value) return updated_value def _container_for(agent: Agent, location: ShareLocation | str) -> MutableMapping[str, Any]: """Return an agent container based on a state-or-memory location.""" normalized = _normalize_location(location) if normalized is ShareLocation.STATE: return _agent_state(agent) if normalized is ShareLocation.MEMORY: return _agent_memory(agent) raise ValueError(f"Unsupported share location: {location!r}") def _iter_world_agents(world: World) -> tuple[Agent, ...]: """Return agents from either dict-backed or sequence-backed worlds.""" agents = getattr(world, "agents", ()) if isinstance(agents, Mapping): values: Iterable[Any] = agents.values() elif isinstance(agents, Iterable) and not isinstance(agents, (str, bytes)): values = agents else: values = () return tuple(agent for agent in values if hasattr(agent, "id")) def _distance_between(left: Agent, right: Agent) -> float | None: """Return Euclidean distance between positioned agents, if comparable.""" left_position = getattr(left, "position", None) right_position = getattr(right, "position", None) if left_position is None or right_position is None: return None try: left_array = np.asarray(left_position, dtype=float) right_array = np.asarray(right_position, dtype=float) except (TypeError, ValueError): return None if left_array.shape != right_array.shape: return None return float(np.linalg.norm(left_array - right_array)) def _relationship_score( agent: Agent, candidate: Agent, relationship_memory_key: str, ) -> float: """Read a deterministic relationship score from agent memory.""" relationships = _agent_memory(agent).get(relationship_memory_key, {}) if not isinstance(relationships, Mapping): return 0.0 entry = relationships.get(_agent_id(candidate), {}) if not isinstance(entry, Mapping): return 0.0 return _as_float(entry.get("score"), 0.0) def _update_relationship( agent: Agent, other: Agent, relationship_memory_key: str, delta: float, step: int | None, interaction: str, ) -> None: """Update relationship memory after a collaboration action.""" memory = _agent_memory(agent) relationships = _ensure_mapping(memory, relationship_memory_key) bucket = _ensure_mapping(relationships, _agent_id(other)) bucket["score"] = _as_float(bucket.get("score"), 0.0) + delta bucket["interaction_count"] = _as_int(bucket.get("interaction_count"), 0) + 1 bucket["last_interaction"] = interaction if step is not None: bucket["last_interaction_step"] = step def _candidate_targets( agent: Agent, world: World, target_types: Sequence[str], max_distance: float | None, ) -> tuple[Agent, ...]: """Return alive candidate targets that satisfy type and distance filters.""" actor_id = _agent_id(agent) allowed_types = {str(target_type) for target_type in target_types} candidates: list[Agent] = [] for candidate in _iter_world_agents(world): if _agent_id(candidate) == actor_id: continue if not _is_alive(candidate): continue if allowed_types and _agent_type(candidate) not in allowed_types: continue distance = _distance_between(agent, candidate) if max_distance is not None and (distance is None or distance > max_distance): continue candidates.append(candidate) return tuple(candidates) def _select_targets( agent: Agent, world: World, *, target_agent_id: str | None, target_types: Sequence[str], max_distance: float | None, max_targets: int, relationship_memory_key: str, ) -> tuple[Agent, ...]: """Select collaboration targets deterministically. Selection first honors an explicit target id. Otherwise it ranks alive candidates by relationship score, spatial distance, and id. """ if max_targets <= 0: return () candidates = _candidate_targets(agent, world, target_types, max_distance) if target_agent_id is not None: target_id = str(target_agent_id) explicit_matches = tuple( candidate for candidate in candidates if _agent_id(candidate) == target_id ) return explicit_matches[:max_targets] def sort_key(candidate: Agent) -> tuple[float, float, str]: score = _relationship_score(agent, candidate, relationship_memory_key) distance = _distance_between(agent, candidate) normalized_distance = float("inf") if distance is None else distance return (-score, normalized_distance, _agent_id(candidate)) return tuple(sorted(candidates, key=sort_key)[:max_targets]) def _first_goal(agent: Agent) -> str | None: """Return a deterministic goal label from an agent, if one exists.""" goals = getattr(agent, "goals", None) if isinstance(goals, Mapping) and goals: return str(sorted(goals.keys(), key=str)[0]) if isinstance(goals, Sequence) and not isinstance(goals, (str, bytes)) and goals: return str(goals[0]) if isinstance(goals, str) and goals: return goals return None def _behavior_names(agent: Agent) -> tuple[str, ...]: """Return stable names for behaviors attached to an agent.""" behaviors = getattr(agent, "behaviors", ()) if isinstance(behaviors, Mapping): raw_behaviors: Iterable[Any] = behaviors.values() elif isinstance(behaviors, Iterable) and not isinstance(behaviors, (str, bytes)): raw_behaviors = behaviors else: raw_behaviors = () names: list[str] = [] for behavior in raw_behaviors: name = getattr(behavior, "name", behavior.__class__.__name__) names.append(str(name)) return tuple(sorted(names)) def _success( behavior: str, agent: Agent, target_ids: Sequence[str] = (), details: Mapping[str, Any] | None = None, world: World | None = None, ) -> dict[str, Any]: """Build a successful behavior outcome dictionary.""" return BehaviorOutcome( behavior=behavior, actor_id=_agent_id(agent), success=True, target_ids=tuple(str(target_id) for target_id in target_ids), step=_world_step(world) if world is not None else None, details=details or {}, ).to_dict() def _failure( behavior: str, agent: Agent, reason: str, details: Mapping[str, Any] | None = None, world: World | None = None, ) -> dict[str, Any]: """Build a failed behavior outcome dictionary.""" payload = {"reason": reason} if details: payload.update(details) logger.debug("Behavior %s failed for agent %s: %s", behavior, _agent_id(agent), reason) return BehaviorOutcome( behavior=behavior, actor_id=_agent_id(agent), success=False, step=_world_step(world) if world is not None else None, details=payload, ).to_dict() @dataclass class CooperateBehavior(Behavior): """Contribute effort toward a shared goal with another agent. The behavior modifies generic state and memory fields only. A farm may use this for cooperative harvesting, a civilization for public works, a lab for joint research, and a fantasy world for party objectives. """ name: ClassVar[str] = "cooperate" target_agent_id: str | None = None target_types: tuple[str, ...] = () max_distance: float | None = None goal: str | None = None effort: float = 1.0 energy_key: str = "energy" energy_cost: float = 0.0 cooperation_state_key: str = "cooperation" shared_goal_memory_key: str = "shared_goals" relationship_memory_key: str = "relationships" relationship_delta: float = 0.1 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can cooperate on this step.""" if not _is_alive(agent) or self.effort <= 0: return False if self.energy_cost > 0: available_energy = _as_float(_get_path(_agent_state(agent), self.energy_key), 0.0) if available_energy < self.energy_cost: return False targets = _select_targets( agent, world, target_agent_id=self.target_agent_id, target_types=self.target_types, max_distance=self.max_distance, max_targets=1, relationship_memory_key=self.relationship_memory_key, ) return bool(targets) def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Execute a cooperative contribution with a selected partner.""" if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) target = _select_targets( agent, world, target_agent_id=self.target_agent_id, target_types=self.target_types, max_distance=self.max_distance, max_targets=1, relationship_memory_key=self.relationship_memory_key, )[0] actor_state = _agent_state(agent) target_state = _agent_state(target) step = _world_step(world) goal_name = self.goal or _first_goal(agent) or _first_goal(target) or "shared" if self.energy_cost > 0: _increment_path(actor_state, self.energy_key, -self.energy_cost) _increment_path(actor_state, self.cooperation_state_key, self.effort) _increment_path(target_state, f"{self.cooperation_state_key}_received", self.effort) actor_shared_goals = _ensure_mapping(_agent_memory(agent), self.shared_goal_memory_key) actor_goal_bucket = _ensure_mapping(actor_shared_goals, goal_name) actor_goal_bucket["contribution"] = ( _as_float(actor_goal_bucket.get("contribution"), 0.0) + self.effort ) actor_partners = _ensure_list(actor_goal_bucket, "partners") _add_unique(actor_partners, _agent_id(target)) target_shared_goals = _ensure_mapping(_agent_memory(target), self.shared_goal_memory_key) target_goal_bucket = _ensure_mapping(target_shared_goals, goal_name) target_goal_bucket["received_contribution"] = ( _as_float(target_goal_bucket.get("received_contribution"), 0.0) + self.effort ) target_partners = _ensure_list(target_goal_bucket, "partners") _add_unique(target_partners, _agent_id(agent)) _update_relationship( agent, target, self.relationship_memory_key, self.relationship_delta, step, self.name, ) _update_relationship( target, agent, self.relationship_memory_key, self.relationship_delta, step, self.name, ) logger.debug( "Agent %s cooperated with %s on goal %s using effort %.3f", _agent_id(agent), _agent_id(target), goal_name, self.effort, ) return _success( self.name, agent, target_ids=(_agent_id(target),), details={ "goal": goal_name, "effort": self.effort, "energy_cost": self.energy_cost, }, world=world, ) @dataclass class CommunicateBehavior(Behavior): """Send a deterministic structured message to one or more agents.""" name: ClassVar[str] = "communicate" target_agent_id: str | None = None target_types: tuple[str, ...] = () max_distance: float | None = None max_recipients: int = 1 message: str = "status_update" channel: str = "default" content_state_keys: tuple[str, ...] = () content_memory_keys: tuple[str, ...] = () inbox_memory_key: str = "inbox" outbox_memory_key: str = "outbox" relationship_memory_key: str = "relationships" relationship_delta: float = 0.05 max_message_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether at least one message recipient is available.""" if not _is_alive(agent) or self.max_recipients <= 0: return False return bool(self._recipients(agent, world)) def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Send the configured message to selected recipients.""" recipients = self._recipients(agent, world) if not _is_alive(agent) or not recipients: return _failure(self.name, agent, "no_available_recipients", world=world) step = _world_step(world) sent_messages: list[dict[str, Any]] = [] actor_outbox = _ensure_list(_agent_memory(agent), self.outbox_memory_key) for recipient in recipients: record = { "from": _agent_id(agent), "to": _agent_id(recipient), "channel": self.channel, "message": self.message, "content": self._message_content(agent), "step": step, } recipient_inbox = _ensure_list(_agent_memory(recipient), self.inbox_memory_key) _append_bounded(recipient_inbox, copy.deepcopy(record), self.max_message_history) _append_bounded(actor_outbox, copy.deepcopy(record), self.max_message_history) _update_relationship( agent, recipient, self.relationship_memory_key, self.relationship_delta, step, self.name, ) _update_relationship( recipient, agent, self.relationship_memory_key, self.relationship_delta, step, self.name, ) sent_messages.append(record) logger.debug( "Agent %s communicated with %s recipient(s) on channel %s", _agent_id(agent), len(recipients), self.channel, ) return _success( self.name, agent, target_ids=tuple(_agent_id(recipient) for recipient in recipients), details={"messages_sent": len(sent_messages), "channel": self.channel}, world=world, ) def _recipients(self, agent: Agent, world: World) -> tuple[Agent, ...]: """Return selected communication recipients.""" return _select_targets( agent, world, target_agent_id=self.target_agent_id, target_types=self.target_types, max_distance=self.max_distance, max_targets=self.max_recipients, relationship_memory_key=self.relationship_memory_key, ) def _message_content(self, agent: Agent) -> dict[str, Any]: """Build a message payload from selected state and memory fields.""" state = _agent_state(agent) memory = _agent_memory(agent) return { "state": { key: copy.deepcopy(_get_path(state, key)) for key in self.content_state_keys }, "memory": { key: copy.deepcopy(_get_path(memory, key)) for key in self.content_memory_keys }, } @dataclass class ShareBehavior(Behavior): """Share a state or memory value with another agent. Numeric values may be transferred or copied. Non-numeric values are copied by default, which makes the behavior appropriate for knowledge, beliefs, messages, plans, map data, or arbitrary DSL-defined concepts. """ name: ClassVar[str] = "share" item_key: str = "shared_value" source: ShareLocation | str = ShareLocation.STATE destination: ShareLocation | str | None = None target_agent_id: str | None = None target_types: tuple[str, ...] = () max_distance: float | None = None amount: float | None = None fraction: float | None = None deplete_source: bool = False allow_non_numeric: bool = True relationship_memory_key: str = "relationships" relationship_delta: float = 0.08 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can share the configured value.""" if not _is_alive(agent) or not _split_path(self.item_key): return False targets = self._targets(agent, world) if not targets: return False try: source_container = _container_for(agent, self.source) except ValueError: return False value = _get_path(source_container, self.item_key) if value is None: return False if _is_number(value): transfer_amount = self._numeric_transfer_amount(float(value)) return transfer_amount > 0 and ( not self.deplete_source or float(value) >= transfer_amount ) return self.allow_non_numeric def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Share the configured value with the selected target.""" if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) target = self._targets(agent, world)[0] step = _world_step(world) source_location = _normalize_location(self.source) destination_location = _normalize_location(self.destination or self.source) source_container = _container_for(agent, source_location) destination_container = _container_for(target, destination_location) value = _get_path(source_container, self.item_key) if _is_number(value): source_value = float(value) transfer_amount = self._numeric_transfer_amount(source_value) destination_value = _as_float(_get_path(destination_container, self.item_key), 0.0) _set_path(destination_container, self.item_key, destination_value + transfer_amount) if self.deplete_source: _set_path(source_container, self.item_key, source_value - transfer_amount) shared_detail: Any = transfer_amount else: copied_value = copy.deepcopy(value) _set_path(destination_container, self.item_key, copied_value) if self.deplete_source: _delete_path(source_container, self.item_key) shared_detail = copied_value _update_relationship( agent, target, self.relationship_memory_key, self.relationship_delta, step, self.name, ) _update_relationship( target, agent, self.relationship_memory_key, self.relationship_delta, step, self.name, ) logger.debug("Agent %s shared %s with %s", _agent_id(agent), self.item_key, _agent_id(target)) return _success( self.name, agent, target_ids=(_agent_id(target),), details={ "item_key": self.item_key, "source": source_location.value, "destination": destination_location.value, "shared_value": copy.deepcopy(shared_detail), "depleted_source": self.deplete_source, }, world=world, ) def _targets(self, agent: Agent, world: World) -> tuple[Agent, ...]: """Return selected share target.""" return _select_targets( agent, world, target_agent_id=self.target_agent_id, target_types=self.target_types, max_distance=self.max_distance, max_targets=1, relationship_memory_key=self.relationship_memory_key, ) def _numeric_transfer_amount(self, available_value: float) -> float: """Resolve a numeric transfer amount from explicit amount or fraction.""" if self.amount is not None: return max(0.0, float(self.amount)) if self.fraction is not None: clipped_fraction = min(max(float(self.fraction), 0.0), 1.0) return available_value * clipped_fraction return available_value if not self.deplete_source else min(1.0, available_value) @dataclass class RecommendBehavior(Behavior): """Recommend an option, behavior, or strategy to another agent. Recommendations are memory records. They do not force the recipient to act, preserving agent autonomy and allowing policies to decide whether to use recommendation memory as context. """ name: ClassVar[str] = "recommend" target_agent_id: str | None = None target_types: tuple[str, ...] = () max_distance: float | None = None recommendation: str | None = None recommendation_type: str = "behavior" confidence: float = 1.0 reason: str = "agent_recommendation" source_memory_key: str = "known_options" recipient_memory_key: str = "recommendations" actor_memory_key: str = "recommendations_made" relationship_memory_key: str = "relationships" relationship_delta: float = 0.04 max_recommendation_history: int = 300 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether a recommendation can be generated and delivered.""" if not _is_alive(agent): return False target = self._target(agent, world) return target is not None and self._resolve_recommendation(agent, target) is not None def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Store a recommendation record in the recipient's memory.""" target = self._target(agent, world) if target is None: return _failure(self.name, agent, "no_available_target", world=world) recommendation = self._resolve_recommendation(agent, target) if recommendation is None: return _failure(self.name, agent, "no_recommendation_available", world=world) step = _world_step(world) record = { "from": _agent_id(agent), "to": _agent_id(target), "type": self.recommendation_type, "recommendation": recommendation, "confidence": min(max(float(self.confidence), 0.0), 1.0), "reason": self.reason, "step": step, } recipient_recommendations = _ensure_list(_agent_memory(target), self.recipient_memory_key) actor_recommendations = _ensure_list(_agent_memory(agent), self.actor_memory_key) _append_bounded( recipient_recommendations, copy.deepcopy(record), self.max_recommendation_history, ) _append_bounded( actor_recommendations, copy.deepcopy(record), self.max_recommendation_history, ) _update_relationship( agent, target, self.relationship_memory_key, self.relationship_delta, step, self.name, ) _update_relationship( target, agent, self.relationship_memory_key, self.relationship_delta, step, self.name, ) logger.debug( "Agent %s recommended %s to %s", _agent_id(agent), recommendation, _agent_id(target), ) return _success( self.name, agent, target_ids=(_agent_id(target),), details={ "recommendation": recommendation, "recommendation_type": self.recommendation_type, "confidence": record["confidence"], }, world=world, ) def _target(self, agent: Agent, world: World) -> Agent | None: """Return the selected recommendation recipient.""" targets = _select_targets( agent, world, target_agent_id=self.target_agent_id, target_types=self.target_types, max_distance=self.max_distance, max_targets=1, relationship_memory_key=self.relationship_memory_key, ) return targets[0] if targets else None def _resolve_recommendation(self, agent: Agent, target: Agent) -> str | None: """Resolve an explicit or inferred recommendation deterministically.""" if self.recommendation: return self.recommendation known_options = _agent_memory(agent).get(self.source_memory_key) if isinstance(known_options, Mapping) and known_options: def option_key(item: tuple[Any, Any]) -> tuple[float, str]: option, score = item return (-_as_float(score, 0.0), str(option)) return str(sorted(known_options.items(), key=option_key)[0][0]) if isinstance(known_options, Sequence) and not isinstance(known_options, (str, bytes)) and known_options: return str(known_options[0]) actor_behavior_names = _behavior_names(agent) target_behavior_names = set(_behavior_names(target)) for behavior_name in actor_behavior_names: if behavior_name not in target_behavior_names: return behavior_name return actor_behavior_names[0] if actor_behavior_names else None @dataclass class NegotiateBehavior(Behavior): """Create or evaluate a structured proposal between two agents. This behavior is intentionally generic. It can represent business deals, social agreements, treaty drafts, marketplace contracts, resource-sharing arrangements, or legal/business negotiation records. It does not encode domain-specific law or economic rules. """ name: ClassVar[str] = "negotiate" target_agent_id: str | None = None target_types: tuple[str, ...] = () max_distance: float | None = None topic: str = "generic" proposal: Mapping[str, Any] = field(default_factory=dict) threshold_state_key: str = "negotiation_threshold" default_threshold: float = 0.0 utility_key: str = "utility" target_utility_key: str = "target_utility" agreement_memory_key: str = "agreements" negotiation_memory_key: str = "negotiations" relationship_memory_key: str = "relationships" accepted_relationship_delta: float = 0.12 proposed_relationship_delta: float = 0.02 max_negotiation_history: int = 300 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether a proposal can be made to a target.""" if not _is_alive(agent) or not self.proposal: return False return self._target(agent, world) is not None def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Evaluate and record a negotiation proposal.""" target = self._target(agent, world) if target is None: return _failure(self.name, agent, "no_available_target", world=world) if not self.proposal: return _failure(self.name, agent, "empty_proposal", world=world) step = _world_step(world) target_utility = self._target_utility() threshold = self._target_threshold(target) accepted = target_utility >= threshold status = "accepted" if accepted else "proposed" agreement_id = self._agreement_id(agent, target, step) record = { "id": agreement_id, "topic": self.topic, "from": _agent_id(agent), "to": _agent_id(target), "proposal": copy.deepcopy(dict(self.proposal)), "target_utility": target_utility, "target_threshold": threshold, "status": status, "step": step, } actor_negotiations = _ensure_list(_agent_memory(agent), self.negotiation_memory_key) target_negotiations = _ensure_list(_agent_memory(target), self.negotiation_memory_key) _append_bounded(actor_negotiations, copy.deepcopy(record), self.max_negotiation_history) _append_bounded(target_negotiations, copy.deepcopy(record), self.max_negotiation_history) relationship_delta = ( self.accepted_relationship_delta if accepted else self.proposed_relationship_delta ) _update_relationship( agent, target, self.relationship_memory_key, relationship_delta, step, self.name, ) _update_relationship( target, agent, self.relationship_memory_key, relationship_delta, step, self.name, ) if accepted: _ensure_mapping(_agent_memory(agent), self.agreement_memory_key)[agreement_id] = copy.deepcopy(record) _ensure_mapping(_agent_memory(target), self.agreement_memory_key)[agreement_id] = copy.deepcopy(record) logger.debug( "Agent %s negotiated with %s on %s: %s", _agent_id(agent), _agent_id(target), self.topic, status, ) return _success( self.name, agent, target_ids=(_agent_id(target),), details={ "agreement_id": agreement_id, "topic": self.topic, "status": status, "target_utility": target_utility, "target_threshold": threshold, }, world=world, ) def _target(self, agent: Agent, world: World) -> Agent | None: """Return the selected negotiation counterparty.""" targets = _select_targets( agent, world, target_agent_id=self.target_agent_id, target_types=self.target_types, max_distance=self.max_distance, max_targets=1, relationship_memory_key=self.relationship_memory_key, ) return targets[0] if targets else None def _target_utility(self) -> float: """Return the proposal utility as seen by the target agent.""" if self.target_utility_key in self.proposal: return _as_float(self.proposal[self.target_utility_key], 0.0) if self.utility_key in self.proposal: return _as_float(self.proposal[self.utility_key], 0.0) return 0.0 def _target_threshold(self, target: Agent) -> float: """Return the target's acceptance threshold from state or memory.""" state_value = _get_path(_agent_state(target), self.threshold_state_key) if _is_number(state_value): return float(state_value) memory_value = _get_path(_agent_memory(target), self.threshold_state_key) if _is_number(memory_value): return float(memory_value) return self.default_threshold def _agreement_id(self, agent: Agent, target: Agent, step: int | None) -> str: """Create a deterministic agreement id for traceability.""" step_label = "unknown_step" if step is None else str(step) return f"{step_label}:{_agent_id(agent)}->{_agent_id(target)}:{self.topic}" Cooperate = CooperateBehavior Communicate = CommunicateBehavior Share = ShareBehavior Recommend = RecommendBehavior Negotiate = NegotiateBehavior BEHAVIOR_REGISTRY: Mapping[str, type[Behavior]] = MappingProxyType( { CooperateBehavior.name: CooperateBehavior, CommunicateBehavior.name: CommunicateBehavior, ShareBehavior.name: ShareBehavior, RecommendBehavior.name: RecommendBehavior, NegotiateBehavior.name: NegotiateBehavior, } ) __all__ = [ "BEHAVIOR_REGISTRY", "BehaviorOutcome", "Communicate", "CommunicateBehavior", "Cooperate", "CooperateBehavior", "Negotiate", "NegotiateBehavior", "Recommend", "RecommendBehavior", "Share", "ShareBehavior", "ShareLocation", ]