defmodule Object.Server do @moduledoc """ GenServer implementation for individual AAOS objects. Each object runs as a separate process with its own state and mailbox, implementing the Actor model with message-passing for true concurrency. ## Features - Individual process per object for isolation and fault tolerance - AI-enhanced reasoning and learning capabilities - Periodic message processing and learning updates - Heartbeat system for dyad maintenance - Integration with schema registry and message routing - Support for meta-DSL constructs and method execution ## Process Lifecycle 1. **Initialization**: Register with schema registry, initialize AI reasoning 2. **Active Phase**: Handle calls/casts, process messages, perform learning 3. **Periodic Tasks**: Heartbeats, message processing, learning updates 4. **Termination**: Cleanup and unregister from schema registry Objects communicate through structured messages routed via the message routing system, with support for interaction dyads and social learning. """ use GenServer require Logger alias Object.{Mailbox, SchemaRegistry, MessageRouter, AIReasoning} # Client API @doc """ Starts a new object server process. Creates a new GenServer process for the object and registers it in the object registry for lookup by ID. The server handles all object lifecycle management including initialization, message processing, and graceful shutdown. ## Parameters - `object_spec` - Object struct containing: - `:id` - Unique object identifier - `:state` - Initial object state - `:methods` - Available methods - `:goal` - Objective function - Additional Object.t() fields ## Returns - `{:ok, pid}` - Success with process PID - `{:error, reason}` - Startup failure reasons: - `:registration_failed` - Object ID already in use - `:init_error` - Initialization failure - `:resource_exhausted` - System resource limits ## Process Lifecycle 1. **Registration**: Register in schema registry and object registry 2. **Initialization**: Set up AI reasoning, health monitoring 3. **Scheduling**: Start periodic tasks (heartbeat, message processing) 4. **Ready**: Process incoming calls and casts ## Examples # Start a basic object server iex> object = Object.new(id: "test_obj", state: %{value: 1}) iex> {:ok, pid} = Object.Server.start_link(object) iex> Process.alive?(pid) true # Start with registry lookup iex> {:ok, _pid} = Object.Server.start_link(object) iex> [{found_pid, _}] = Registry.lookup(Object.Registry, "test_obj") iex> Process.alive?(found_pid) true ## Error Handling - Duplicate object IDs are rejected - Initialization failures are logged and reported - Resource exhaustion triggers graceful degradation - Process crashes are handled by the supervisor ## Performance - Startup time: ~5-10ms including registration - Memory usage: ~2KB base + object state size - Registry lookup: O(1) by object ID """ @spec start_link(Object.t()) :: GenServer.on_start() def start_link(object_spec) do object_id = object_spec.id GenServer.start_link(__MODULE__, object_spec, name: via_registry(object_id)) end @doc """ Sends a message from one object to another. Routes a message through the message routing system with specified type, content, and delivery options. Messages are processed asynchronously with comprehensive error handling and delivery guarantees. ## Parameters - `object_id` - Sender object ID (must be registered) - `to_object_id` - Recipient object ID - `message_type` - Type of message for routing and handling: - `:state_update` - State change notification - `:coordination` - Coordination request/response - `:learning_signal` - Learning data or feedback - `:heartbeat` - Connection health check - `:negotiation` - Multi-step negotiation protocol - `content` - Message content (any serializable term) - `opts` - Delivery options: - `:priority` - `:low | :medium | :high | :critical` (default: `:medium`) - `:ttl` - Time-to-live in seconds (default: 3600) - `:requires_ack` - Delivery confirmation required (default: false) - `:retry_count` - Maximum retry attempts (default: 3) ## Returns - `:ok` - Message sent successfully - `{:error, reason}` - Send failure: - `:object_not_found` - Sender object not registered - `:throttled` - System under load, message delayed - `:invalid_recipient` - Recipient object invalid - `:message_too_large` - Content exceeds size limits ## Examples # Send coordination request iex> Object.Server.send_message( ...> "agent_1", "coordinator_1", :coordination, ...> %{action: :join_coalition, priority: :high}, ...> [priority: :high, requires_ack: true] ...> ) :ok # Send learning signal with TTL iex> Object.Server.send_message( ...> "learner_1", "teacher_1", :learning_signal, ...> %{reward: 1.0, experience: %{action: :explore}}, ...> [ttl: 300, priority: :medium] ...> ) :ok # Error: Object not found iex> Object.Server.send_message( ...> "nonexistent", "agent_2", :hello, %{} ...> ) {:error, :object_not_found} ## Delivery Guarantees - **Best Effort**: Default delivery without acknowledgment - **At Least Once**: With `requires_ack: true` - **Circuit Breaker**: Prevents cascade failures - **Dead Letter Queue**: Failed messages queued for retry ## Performance - Message latency: ~1-5ms for local objects - Throughput: ~1000 messages/sec per object - Backpressure: Automatic throttling under load - Resource usage: ~100 bytes per message overhead """ @spec send_message(Object.object_id(), Object.object_id(), atom(), any(), keyword()) :: :ok | {:error, atom()} def send_message(object_id, to_object_id, message_type, content, opts \\ []) do GenServer.call(via_registry(object_id), {:send_message, to_object_id, message_type, content, opts}) end @doc """ Gets the current state of an object. Retrieves the complete current state of the specified object. This is a synchronous operation that returns the state at the time of the call. ## Parameters - `object_id` - Object identifier (must be registered) ## Returns - Object's current state map - May raise if object not found or unresponsive ## Examples # Get state of a sensor object iex> state = Object.Server.get_state("temp_sensor_1") iex> state.temperature 22.5 # Check AI agent state iex> Object.Server.get_state("ai_agent_alpha") %{intelligence_level: :advanced, current_task: :learning} ## Error Handling - Raises `:noproc` if object process not found - Raises `:timeout` if object unresponsive (default: 5 seconds) - State is read-only snapshot at call time ## Performance - Operation time: ~0.1-1ms for local objects - No side effects on object state - Thread-safe read operation """ @spec get_state(Object.object_id()) :: Object.object_state() def get_state(object_id) do GenServer.call(via_registry(object_id), :get_state) end @doc """ Updates an object's internal state. Merges the provided updates into the object's current state with validation and error handling. The update is atomic and will either completely succeed or leave the state unchanged. ## Parameters - `object_id` - Object identifier (must be registered) - `state_updates` - Map of state updates to merge with current state ## Returns - `:ok` - Update successful - `{:error, reason}` - Update failed: - `:object_not_found` - Object not registered - `:invalid_state` - State update validation failed - `:state_too_large` - Update would exceed size limits - `:forbidden_keys` - Update contains forbidden keys ## Validation Rules - State updates must be maps - Combined state cannot exceed 100 entries - Cannot update system keys (:__internal__, :__system__, :__meta__) - Numeric values must be finite numbers ## Examples # Update sensor readings iex> Object.Server.update_state("temp_sensor_1", %{ ...> temperature: 23.5, ...> humidity: 68, ...> last_reading: DateTime.utc_now() ...> }) :ok # Incremental energy update iex> Object.Server.update_state("robot_1", %{energy: 95}) :ok # Error: Forbidden key iex> Object.Server.update_state("agent_1", %{__system__: "hack"}) {:error, :forbidden_keys} ## Atomicity - Updates are applied atomically (all or nothing) - Concurrent updates are serialized - No partial state corruption possible - Original state preserved on validation failure ## Performance - Update time: ~0.5-2ms depending on state size - Memory overhead: Temporary copy during validation - Throughput: ~500 updates/sec per object """ @spec update_state(Object.object_id(), Object.object_state()) :: :ok | {:error, atom()} def update_state(object_id, state_updates) do GenServer.call(via_registry(object_id), {:update_state, state_updates}) end @doc """ Executes a method on an object. Calls the specified method with given arguments on the target object with comprehensive error handling, resource protection, and performance monitoring. Methods are executed within the object's process context. ## Parameters - `object_id` - Object identifier (must be registered) - `method` - Method name (atom, must be in object's methods list) - `args` - Method arguments list (default: []) ## Returns - `:ok` - Method executed successfully - `{:error, reason}` - Execution failed: - `:object_not_found` - Object not registered - `:method_not_available` - Method not in object's methods list - `:timeout` - Method execution exceeded time limit - `:resource_exhausted` - Insufficient system resources - `:method_error` - Method execution failed internally ## Built-in Methods Common methods available on most objects: - `:update_state` - Update internal state - `:interact` - Process interaction with environment - `:learn` - Apply learning from experience - `:evaluate_goal` - Compute goal satisfaction Subtype-specific methods: - AI Agents: `:reason`, `:plan`, `:adapt` - Sensors: `:sense`, `:calibrate`, `:filter_noise` - Actuators: `:execute_action`, `:queue_action` - Coordinators: `:coordinate`, `:allocate_resources` ## Examples # Execute learning method iex> experience = %{reward: 1.0, action: :explore, state: %{x: 1}} iex> Object.Server.execute_method("agent_1", :learn, [experience]) :ok # Execute sensor calibration iex> Object.Server.execute_method("temp_sensor_1", :calibrate, []) :ok # Error: Method not available iex> Object.Server.execute_method("sensor_1", :fly, []) {:error, :method_not_available} ## Resource Protection - Resource permission checked before execution - Circuit breakers prevent system overload - Execution timeouts prevent runaway methods - Performance monitoring tracks method efficiency ## Performance - Method latency: ~1-10ms depending on complexity - Timeout: 30 seconds default - Retry logic: 3 attempts with exponential backoff - Monitoring: Execution time and success rate tracked """ @spec execute_method(Object.object_id(), Object.method_name(), [any()]) :: :ok | {:error, atom()} def execute_method(object_id, method, args) do GenServer.call(via_registry(object_id), {:execute_method, method, args}) end @doc """ Applies a meta-DSL construct to an object. Executes meta-language constructs for self-reflection and modification, enabling objects to reason about and modify their own behavior, goals, and knowledge structures. ## Parameters - `object_id` - Object identifier (must be registered) - `construct` - Meta-DSL construct to execute: - `:define` - Define new attributes or capabilities - `:goal` - Modify objective functions - `:belief` - Update world model beliefs - `:infer` - Perform inference on current knowledge - `:decide` - Make decisions based on current state - `:learn` - Process learning experiences - `:refine` - Adjust learning parameters - `args` - Arguments specific to the construct ## Returns Result map containing updates to be applied: - `:state_updates` - Updates to object state - `:world_model_updates` - Updates to world model - `:goal_update` - New goal function - `:meta_dsl_updates` - Updates to meta-DSL state ## Meta-DSL Constructs ### DEFINE - Create New Capabilities ```elixir Object.Server.apply_meta_dsl("agent_1", :define, {:confidence, 0.8}) # => %{state_updates: %{confidence: 0.8}} ``` ### INFER - Bayesian Reasoning ```elixir inference_data = %{observations: [%{light: :on}], priors: %{light: :off}} Object.Server.apply_meta_dsl("agent_1", :infer, inference_data) # => %{world_model_updates: %{beliefs: %{light: :on}}} ``` ### DECIDE - Goal-Directed Choice ```elixir context = %{options: [:explore, :exploit], current_reward: 0.5} Object.Server.apply_meta_dsl("agent_1", :decide, context) # => %{state_updates: %{last_action: :explore}} ``` ### REFINE - Meta-Learning ```elixir refinement = %{performance: 0.8, learning_rate: 0.01} Object.Server.apply_meta_dsl("agent_1", :refine, refinement) # => %{meta_dsl_updates: %{learning_parameters: %{...}}} ``` ## Examples # Define new attribute iex> result = Object.Server.apply_meta_dsl("agent_1", :define, {:trust, 0.9}) iex> result.state_updates %{trust: 0.9} # Update beliefs through inference iex> inference = %{evidence: %{sensor_reading: 22.5}} iex> Object.Server.apply_meta_dsl("agent_1", :infer, inference) %{world_model_updates: %{beliefs: %{temperature: 22.5}}} ## Self-Modification Meta-DSL enables powerful self-modification: - **Behavioral Adaptation**: Change response patterns - **Goal Evolution**: Modify objectives based on experience - **Knowledge Integration**: Update beliefs through reasoning - **Parameter Tuning**: Optimize learning parameters ## Safety - Construct validation prevents invalid modifications - Bounded update sizes prevent resource exhaustion - Rollback capability for failed modifications - Audit trail for all self-modifications """ @spec apply_meta_dsl(Object.object_id(), atom(), any()) :: %{ optional(:state_updates) => map(), optional(:world_model_updates) => map(), optional(:goal_update) => Object.goal_function(), optional(:meta_dsl_updates) => map() } def apply_meta_dsl(object_id, construct, args) do GenServer.call(via_registry(object_id), {:apply_meta_dsl, construct, args}) end @doc """ Forms an interaction dyad with another object. Creates a bidirectional communication relationship between two objects with the specified compatibility score. Dyads enable enhanced communication, coordination, and social learning between object pairs. ## Parameters - `object_id` - First object ID (must be registered) - `other_object_id` - Second object ID (must be registered) - `compatibility_score` - Initial compatibility assessment (0.0-1.0, default 0.5) ## Returns - `:ok` - Dyad formed successfully - `{:error, reason}` - Formation failed: - `:object_not_found` - One or both objects not registered - `:self_dyad_not_allowed` - Cannot form dyad with self - `:dyad_already_exists` - Dyad already established - `:compatibility_too_low` - Compatibility below minimum threshold ## Compatibility Guidelines - `0.0-0.3` - Low compatibility, limited interaction benefit - `0.3-0.7` - Moderate compatibility, good for specific tasks - `0.7-1.0` - High compatibility, excellent collaboration potential ## Examples # Form high-compatibility dyad iex> Object.Server.form_dyad("ai_agent_1", "ai_agent_2", 0.8) :ok # Form sensor-coordinator dyad iex> Object.Server.form_dyad("temp_sensor_1", "coordinator_1", 0.6) :ok # Error: Self-dyad not allowed iex> Object.Server.form_dyad("agent_1", "agent_1") {:error, :self_dyad_not_allowed} ## Dyad Benefits - **Priority Messaging**: Dyad partners get message priority - **Social Learning**: Shared experiences and knowledge transfer - **Coordination**: Simplified cooperation protocols - **Trust Building**: Reputation and reliability tracking - **Performance**: Reduced coordination overhead ## Lifecycle Management Dyads evolve over time: 1. **Formation**: Initial creation with compatibility score 2. **Interaction**: Regular communication and coordination 3. **Adaptation**: Compatibility adjustment based on outcomes 4. **Maintenance**: Heartbeat and health monitoring 5. **Dissolution**: Automatic or manual termination ## Performance - Formation time: ~1-3ms - Memory overhead: ~200 bytes per dyad - Maximum dyads per object: 50 (configurable) - Automatic cleanup of inactive dyads """ @spec form_dyad(Object.object_id(), Object.object_id(), float()) :: :ok | {:error, atom()} def form_dyad(object_id, other_object_id, compatibility_score \\ 0.5) do GenServer.call(via_registry(object_id), {:form_dyad, other_object_id, compatibility_score}) end @doc """ Gets communication statistics for an object. Returns comprehensive metrics about the object's messaging patterns, interaction history, and communication performance. Useful for monitoring, debugging, and performance optimization. ## Parameters - `object_id` - Object identifier (must be registered) ## Returns Statistics map containing: - `:total_messages_sent` - Number of messages sent by this object - `:total_messages_received` - Number of messages received - `:pending_inbox` - Current unprocessed inbox messages - `:pending_outbox` - Current unsent outbox messages - `:active_dyads` - Number of currently active interaction dyads - `:total_dyads` - Total dyads ever formed (including inactive) - `:history_size` - Current message history size - `:uptime` - Object uptime in seconds since creation - `:message_rate` - Messages per second (sent + received) - `:dyad_efficiency` - Ratio of active to total dyads ## Examples # Get basic communication stats iex> stats = Object.Server.get_stats("agent_1") iex> stats.total_messages_sent 42 iex> stats.active_dyads 3 # Calculate performance metrics iex> stats = Object.Server.get_stats("coordinator_1") iex> stats.message_rate 2.5 # messages per second iex> stats.dyad_efficiency 0.75 # 75% of dyads are active ## Performance Indicators Key metrics for monitoring: ### Communication Volume - **High volume** (>100 msg/min): Active coordinator or hub object - **Medium volume** (10-100 msg/min): Regular operational object - **Low volume** (<10 msg/min): Peripheral or specialized object ### Dyad Health - **High efficiency** (>0.8): Well-connected, active collaboration - **Medium efficiency** (0.5-0.8): Selective partnerships - **Low efficiency** (<0.5): Poor partner selection or inactive ### Processing Performance - **Low pending**: Efficient message processing - **High pending**: Processing bottleneck or overload ## Monitoring Uses - **Performance Tuning**: Identify communication bottlenecks - **Health Monitoring**: Detect failed or overloaded objects - **Social Analysis**: Understand interaction patterns - **Debugging**: Trace message flow and delivery issues - **Capacity Planning**: Plan for system scaling ## Historical Trends Statistics can be sampled over time to identify: - Communication pattern changes - Performance degradation - Social network evolution - Seasonal or cyclical behaviors """ @spec get_stats(Object.object_id()) :: %{ total_messages_sent: non_neg_integer(), total_messages_received: non_neg_integer(), pending_inbox: non_neg_integer(), pending_outbox: non_neg_integer(), active_dyads: non_neg_integer(), total_dyads: non_neg_integer(), history_size: non_neg_integer(), uptime: non_neg_integer(), message_rate: float(), dyad_efficiency: float() } def get_stats(object_id) do GenServer.call(via_registry(object_id), :get_stats) end # Server callbacks @impl true def init(object_spec) do try do # Register object in schema registry with error handling case SchemaRegistry.register_object(object_spec) do :ok -> Logger.debug("Object #{object_spec.id} registered in schema registry") {:error, reason} -> Logger.error("Failed to register object #{object_spec.id}: #{inspect(reason)}") {:stop, {:registration_failed, reason}} end # Register with health monitor register_with_health_monitor(object_spec) # Initialize AI reasoning capabilities if enabled case AIReasoning.initialize_object_reasoning(object_spec.id) do {:ok, _} -> Logger.info("AI reasoning initialized for object #{object_spec.id}") {:error, reason} -> Logger.warning("Failed to initialize AI reasoning: #{inspect(reason)}") # Continue without AI reasoning end # Start periodic tasks with error protection schedule_heartbeat() schedule_message_processing() schedule_learning_update() # Set up process monitoring Process.flag(:trap_exit, true) Logger.info("Object #{object_spec.id} started successfully") {:ok, Map.put(object_spec, :start_time, System.monotonic_time(:second))} rescue error -> Logger.error("Failed to initialize object #{object_spec.id}: #{inspect(error)}") {:stop, {:init_error, error}} end end @impl true def handle_call({:send_message, to_object_id, message_type, content, opts}, _from, object) do try do # Skip resource permission check updated_object = Object.send_message(object, to_object_id, message_type, content, opts) # Route message through message router with error handling message = create_message(object.id, to_object_id, message_type, content, opts) case MessageRouter.route_message(message) do :ok -> {:reply, :ok, updated_object} {:error, reason} -> Logger.warning("Message routing failed: #{inspect(reason)}") {:reply, {:error, reason}, object} end rescue error -> Logger.error("Critical error in handle_call for object #{object.id}: #{inspect(error)}") {:reply, {:error, {:exception, error}}, object} end end @impl true def handle_call(:get_state, _from, object) do {:reply, object.state, object} end @impl true def handle_call({:update_state, state_updates}, _from, object) do updated_object = Object.update_state(object, state_updates) {:reply, :ok, updated_object} end @impl true def handle_call({:execute_method, method, args}, _from, object) do try do # Enhanced method execution with comprehensive error handling case Object.execute_method(object, method, args) do {:ok, updated_object} -> {:reply, :ok, updated_object} {:error, reason} -> # Log method execution failure Logger.warning("Method execution failed for object #{object.id}: #{method} - #{inspect(reason)}") # Check if this is a critical failure that requires process restart case reason do {:exception, _} -> # Critical failure logged Logger.error("Critical method failure for object #{object.id}, method #{method}: #{inspect(reason)}") _ -> # Non-critical failure logged Logger.warning("Method failure for object #{object.id}, method #{method}: #{inspect(reason)}") end {:reply, {:error, reason}, object} end rescue error -> Logger.error("Critical error in method execution for object #{object.id}: #{inspect(error)}") {:reply, {:error, {:server_exception, error}}, object} end end @impl true def handle_call({:apply_meta_dsl, construct, args}, _from, object) do result = Object.apply_meta_dsl(object, construct, args) # Apply updates to object state updated_object = case result do %{state_updates: updates} when map_size(updates) > 0 -> Object.update_state(object, updates) %{world_model_updates: updates} when map_size(updates) > 0 -> Object.update_world_model(object, updates) _ -> object end {:reply, result, updated_object} end @impl true def handle_call({:form_dyad, other_object_id, compatibility_score}, _from, object) do updated_object = Object.form_interaction_dyad(object, other_object_id, compatibility_score) {:reply, :ok, updated_object} end @impl true def handle_call(:get_stats, _from, object) do stats = Object.get_communication_stats(object) {:reply, stats, object} end def handle_call(:health_check, _from, object) do health_status = check_object_health_internal(object) {:reply, health_status, object} end @impl true def handle_cast({:receive_message, message}, object) do try do # Process message directly without resource monitoring process_incoming_message(object, message) rescue error -> Logger.error("Critical error in handle_cast for object #{object.id}: #{inspect(error)}") {:noreply, object} end end @impl true def handle_cast({:process_messages}, object) do {_processed_messages, updated_object} = Object.process_messages(object) {:noreply, updated_object} end @impl true def handle_info(:heartbeat, object) do # Send heartbeat to connected objects active_dyads = Mailbox.get_active_dyads(object.mailbox) Enum.each(active_dyads, fn {_dyad_id, dyad} -> {from_id, to_id} = dyad.participants other_id = if from_id == object.id, do: to_id, else: from_id heartbeat_message = create_heartbeat_message(object.id, other_id) MessageRouter.route_message(heartbeat_message) end) schedule_heartbeat() {:noreply, object} end @impl true def handle_info(:process_messages, object) do {_processed_messages, updated_object} = Object.process_messages(object) schedule_message_processing() {:noreply, updated_object} end @impl true def handle_info(:learning_update, object) do # Perform AI-enhanced learning update based on recent experiences experience = extract_recent_experience(object) case AIReasoning.synthesize_learning(object.id, experience.recent_interactions, object.state, experience.performance_metrics, extract_environmental_context(object)) do {:ok, learning_insights} -> Logger.debug("Learning insights for #{object.id}: #{inspect(learning_insights)}") # Apply AI-synthesized learning enhanced_experience = Map.put(experience, :ai_insights, learning_insights) updated_object = Object.learn(object, enhanced_experience) schedule_learning_update() {:noreply, updated_object} {:error, _reason} -> # Fallback to standard learning without AI enhancement updated_object = Object.learn(object, experience) schedule_learning_update() {:noreply, updated_object} end end @impl true def handle_info({:DOWN, _ref, :process, _pid, reason}, object) do Logger.warning("Connected process died: #{inspect(reason)}") {:noreply, object} end def handle_info({:delayed_message, message}, object) do Logger.debug("Processing delayed message for object #{object.id}") handle_cast({:receive_message, message}, object) end @impl true def terminate(reason, object) do Logger.info("Object #{object.id} terminating: #{inspect(reason)}") try do # Perform graceful cleanup cleanup_object_resources(object) # Unregister from schema registry case SchemaRegistry.unregister_object(object.id) do :ok -> Logger.debug("Object #{object.id} unregistered successfully") {:error, unregister_reason} -> Logger.warning("Failed to unregister object #{object.id}: #{inspect(unregister_reason)}") end # Health monitoring disabled - Object.HealthMonitor not available :ok # Report termination reason if abnormal case reason do :normal -> :ok :shutdown -> :ok {:shutdown, _} -> :ok _ -> Logger.error("Object #{object.id} terminated abnormally: #{inspect(reason)}") end rescue cleanup_error -> Logger.error("Error during object #{object.id} cleanup: #{inspect(cleanup_error)}") end :ok end # Private functions defp via_registry(object_id) do {:via, Registry, {Object.Registry, object_id}} end defp schedule_heartbeat do Process.send_after(self(), :heartbeat, 5_000) # 5 seconds end defp schedule_message_processing do Process.send_after(self(), :process_messages, 1_000) # 1 second end defp schedule_learning_update do Process.send_after(self(), :learning_update, 10_000) # 10 seconds end defp create_message(from_id, to_id, message_type, content, opts) do %{ id: generate_message_id(), from: from_id, to: to_id, type: message_type, content: content, timestamp: DateTime.utc_now(), priority: Keyword.get(opts, :priority, :medium), requires_ack: Keyword.get(opts, :requires_ack, false), ttl: Keyword.get(opts, :ttl, 3600) } end defp create_heartbeat_message(from_id, to_id) do %{ id: generate_message_id(), from: from_id, to: to_id, type: :heartbeat, content: %{timestamp: DateTime.utc_now()}, timestamp: DateTime.utc_now(), priority: :low, requires_ack: false, ttl: 30 } end defp generate_message_id do :crypto.strong_rand_bytes(8) |> Base.encode16() |> String.downcase() end defp extract_recent_experience(object) do # Extract recent interactions for learning recent_messages = object.mailbox.message_history |> Enum.take(10) |> Enum.map(fn msg -> %{type: msg.type, reward: calculate_message_reward(msg)} end) %{ recent_interactions: recent_messages, performance_metrics: calculate_performance_metrics(object), timestamp: DateTime.utc_now() } end defp calculate_message_reward(_message) do # Simplified reward calculation :rand.uniform() end defp calculate_performance_metrics(object) do stats = Object.get_communication_stats(object) %{ message_efficiency: stats.total_messages_sent / max(1, stats.uptime), dyad_health: length(Mailbox.get_active_dyads(object.mailbox)) / max(1, stats.total_dyads), goal_achievement: Object.evaluate_goal(object) } end defp extract_message_context(object, message) do %{ current_state: object.state, recent_interactions: object.mailbox.message_history |> Enum.take(5), active_dyads: Mailbox.get_active_dyads(object.mailbox), message_timestamp: message.timestamp, message_priority: message.priority } end defp extract_environmental_context(object) do %{ active_objects: Registry.select(Object.Registry, [{{:"$1", :"$2", :"$3"}, [], [:"$1"]}]) |> length(), system_load: :erlang.statistics(:scheduler_utilization), current_time: DateTime.utc_now(), object_uptime: System.monotonic_time(:second) - Map.get(object, :start_time, 0) } end # Enhanced error handling support functions defp register_with_health_monitor(object_spec) do try do # Health monitoring disabled - Object.HealthMonitor not available :ok Logger.debug("Object #{object_spec.id} registered with health monitor") rescue error -> Logger.warning("Failed to register object #{object_spec.id} with health monitor: #{inspect(error)}") end end defp process_incoming_message(object, message) do # Use AI reasoning to analyze incoming message if available case AIReasoning.analyze_message(object.id, message.from, message.content, extract_message_context(object, message)) do {:ok, analysis} -> Logger.debug("Message analysis for #{object.id}: #{inspect(analysis)}") # Process message with AI insights case Object.receive_message(object, message) do {:ok, updated_object} -> {:noreply, updated_object} {:error, reason} -> Logger.warning("Failed to receive message: #{inspect(reason)}") handle_message_failure(object, message, reason) end {:error, _reason} -> # Fallback to standard message processing without AI analysis case Object.receive_message(object, message) do {:ok, updated_object} -> {:noreply, updated_object} {:error, reason} -> Logger.warning("Failed to receive message: #{inspect(reason)}") handle_message_failure(object, message, reason) end end end defp handle_message_failure(object, message, reason) do # Simple error handling without dead letter queue case reason do {:timeout, _} -> # Retry the message after a delay retry_delay = calculate_retry_delay(reason) Process.send_after(self(), {:delayed_message, message}, retry_delay) Logger.debug("Scheduling message retry for object #{object.id} in #{retry_delay}ms") {:noreply, object} _ -> # Log error and continue Logger.warning("Message processing failed for object #{object.id}: #{inspect(reason)}") {:noreply, object} end end defp calculate_retry_delay(reason) do base_delay = case reason do {:timeout, _} -> 1000 # 1 second for timeouts {:mailbox_full, _} -> 5000 # 5 seconds for mailbox issues _ -> 2000 # 2 seconds default end # Add jitter jitter = :rand.uniform(trunc(base_delay * 0.2)) base_delay + jitter end defp cleanup_object_resources(object) do try do # Clean up any object-specific resources cleanup_tasks = [ fn -> cleanup_mailbox_resources(object) end, fn -> cleanup_interaction_history(object) end, fn -> cleanup_learning_state(object) end ] results = Enum.map(cleanup_tasks, fn task -> try do task.() :ok rescue error -> {:error, error} end end) failed_cleanups = Enum.filter(results, &(elem(&1, 0) == :error)) if not Enum.empty?(failed_cleanups) do Logger.warning("Some cleanup tasks failed for object #{object.id}: #{inspect(failed_cleanups)}") end rescue error -> Logger.error("Error during resource cleanup for object #{object.id}: #{inspect(error)}") end end defp cleanup_mailbox_resources(object) do # Clean up mailbox resources Logger.debug("Cleaning up mailbox resources for object #{object.id}") # Implementation would clean up message queues, connections, etc. :ok end defp cleanup_interaction_history(object) do # Clean up interaction history if it's too large if length(object.interaction_history) > 1000 do Logger.debug("Cleaning up interaction history for object #{object.id}") # Could archive or truncate history end :ok end defp cleanup_learning_state(object) do # Clean up learning state Logger.debug("Cleaning up learning state for object #{object.id}") # Implementation would clean up learning artifacts, cached models, etc. :ok end # Additional handle_info implementations defp check_object_health_internal(object) do try do # Check various health indicators health_indicators = [ check_mailbox_health(object), check_interaction_health(object), check_learning_health(object), check_resource_usage(object) ] failed_indicators = Enum.filter(health_indicators, &(&1 != :ok)) case length(failed_indicators) do 0 -> :ok count when count <= 1 -> :degraded _ -> :failing end rescue _ -> :failing end end defp check_mailbox_health(object) do # Check if mailbox is functioning properly if map_size(object.mailbox.message_history) > 10000 do :mailbox_overloaded else :ok end end defp check_interaction_health(object) do # Check interaction patterns recent_interactions = Enum.take(object.interaction_history, 10) failed_interactions = Enum.count(recent_interactions, fn interaction -> case interaction.outcome do {:error, _} -> true _ -> false end end) if failed_interactions > 5 do :high_failure_rate else :ok end end defp check_learning_health(_object) do # Check learning system health # This would check if learning is progressing normally :ok end defp check_resource_usage(_object) do # Check object-specific resource usage # This would monitor memory usage, computation time, etc. :ok end end