hippocampaif/core_knowledge/agent_system.py

"""
Agent System — Core Knowledge of Agents

Infants distinguish agents from objects by detecting:
1. Self-propulsion: Agents can initiate motion without external contact
2. Goal-directedness: Agents take efficient paths toward goals
3. Contingency: Agents respond to other agents' behavior

This module provides innate priors for identifying and reasoning about 
intentional agents in the environment.

Author: Algorembrant, Rembrant Oyangoren Albeos (2026)
"""

import numpy as np
from typing import Optional


class TrackedAgent:
    """An entity being evaluated for agency."""
    
    __slots__ = ['agent_id', 'position_history', 'agency_score',
                 'self_propelled', 'goal_position', 'efficiency_history']
    
    def __init__(self, agent_id: int):
        self.agent_id = agent_id
        self.position_history: list[np.ndarray] = []
        self.agency_score = 0.0  # 0 = object, 1 = definitely an agent
        self.self_propelled = False
        self.goal_position: Optional[np.ndarray] = None
        self.efficiency_history: list[float] = []


class AgentSystem:
    """
    Innate agency detection system.
    
    Evaluates whether tracked entities are intentional agents based on
    three core cues: self-propulsion, goal-directedness, and contingency.
    This is an innate prior — infants as young as 3 months make these
    distinctions.
    """
    
    def __init__(self, 
                 self_propulsion_threshold: float = 0.3,
                 efficiency_threshold: float = 0.6,
                 history_window: int = 20):
        """
        Args:
            self_propulsion_threshold: Min velocity change without contact to flag self-propulsion.
            efficiency_threshold: Min path efficiency to flag goal-directedness.
            history_window: Number of frames to consider for agency evaluation.
        """
        self.self_propulsion_threshold = self_propulsion_threshold
        self.efficiency_threshold = efficiency_threshold
        self.history_window = history_window
        self.agents: dict[int, TrackedAgent] = {}
        
    def update_entity(self, entity_id: int, position: np.ndarray,
                      was_contacted: bool = False):
        """
        Update an entity's trajectory and evaluate agency cues.
        
        Args:
            entity_id: Unique identifier for this entity.
            position: Current position.
            was_contacted: Whether another object contacted this entity this frame.
        """
        if entity_id not in self.agents:
            self.agents[entity_id] = TrackedAgent(entity_id)
            
        agent = self.agents[entity_id]
        pos = np.asarray(position, dtype=np.float64)
        agent.position_history.append(pos)
        
        # Trim history
        if len(agent.position_history) > self.history_window:
            agent.position_history = agent.position_history[-self.history_window:]
        
        # --- CUE 1: Self-propulsion ---
        if len(agent.position_history) >= 3:
            # Velocity change without external contact = self-propulsion
            v_prev = agent.position_history[-2] - agent.position_history[-3]
            v_curr = agent.position_history[-1] - agent.position_history[-2]
            accel = np.linalg.norm(v_curr - v_prev)
            
            if accel > self.self_propulsion_threshold and not was_contacted:
                agent.self_propelled = True
        
        # --- CUE 2: Goal-directedness ---
        self._evaluate_goal_directedness(agent)
        
        # --- Compute composite agency score ---
        self._compute_agency_score(agent)
    
    def _evaluate_goal_directedness(self, agent: TrackedAgent):
        """
        Evaluate whether the entity takes efficient paths toward a goal.
        
        Efficiency = direct_distance / path_length
        Agents take short, efficient paths. Objects follow ballistic arcs.
        """
        if len(agent.position_history) < 5:
            return
            
        # Use the last position as the "observed goal"
        start = agent.position_history[0]
        end = agent.position_history[-1]
        
        direct_dist = np.linalg.norm(end - start)
        if direct_dist < 0.01:
            return  # Stationary
            
        # Compute path length
        path_length = 0.0
        for i in range(1, len(agent.position_history)):
            path_length += np.linalg.norm(
                agent.position_history[i] - agent.position_history[i-1]
            )
        
        if path_length < 0.01:
            return
            
        efficiency = direct_dist / path_length  # 1.0 = perfectly direct
        agent.efficiency_history.append(efficiency)
        
        # Trim
        if len(agent.efficiency_history) > 10:
            agent.efficiency_history = agent.efficiency_history[-10:]
    
    def _compute_agency_score(self, agent: TrackedAgent):
        """Combine cues into a single agency belief."""
        score = 0.0
        
        # Self-propulsion is a strong cue
        if agent.self_propelled:
            score += 0.5
        
        # Goal-directedness
        if agent.efficiency_history:
            avg_eff = np.mean(agent.efficiency_history)
            if avg_eff > self.efficiency_threshold:
                score += 0.3
            else:
                score += 0.1 * avg_eff
        
        # Motion variability (agents move more erratically than ballistic objects)
        if len(agent.position_history) >= 3:
            velocities = []
            for i in range(1, len(agent.position_history)):
                v = agent.position_history[i] - agent.position_history[i-1]
                velocities.append(v)
            if len(velocities) >= 2:
                vel_array = np.array(velocities)
                direction_changes = 0
                for i in range(1, len(vel_array)):
                    dot = np.dot(vel_array[i], vel_array[i-1])
                    if dot < 0:  # Direction reversal
                        direction_changes += 1
                variability = direction_changes / len(vel_array)
                score += 0.2 * variability
        
        agent.agency_score = np.clip(score, 0.0, 1.0)
    
    def is_agent(self, entity_id: int) -> bool:
        """Check if an entity is believed to be an intentional agent."""
        agent = self.agents.get(entity_id)
        if agent is None:
            return False
        return agent.agency_score > 0.5
    
    def get_agency_score(self, entity_id: int) -> float:
        """Get the agency belief for an entity (0=object, 1=agent)."""
        agent = self.agents.get(entity_id)
        if agent is None:
            return 0.0
        return agent.agency_score
    
    def evaluate_contingency(self, id_a: int, id_b: int) -> float:
        """
        Evaluate contingency between two entities.
        
        Contingency = one entity's actions correlate with another's.
        This is a strong cue for social interaction.
        
        Returns:
            Contingency score (0 to 1).
        """
        a = self.agents.get(id_a)
        b = self.agents.get(id_b)
        if a is None or b is None:
            return 0.0
        
        min_len = min(len(a.position_history), len(b.position_history))
        if min_len < 3:
            return 0.0
        
        # Compute velocity correlation
        va = np.diff(np.array(a.position_history[-min_len:]), axis=0)
        vb = np.diff(np.array(b.position_history[-min_len:]), axis=0)
        
        # Cross-correlation of velocity magnitudes
        mag_a = np.linalg.norm(va, axis=1)
        mag_b = np.linalg.norm(vb, axis=1)
        
        if np.std(mag_a) < 1e-8 or np.std(mag_b) < 1e-8:
            return 0.0
        
        correlation = np.corrcoef(mag_a, mag_b)[0, 1]
        return float(np.clip(abs(correlation), 0.0, 1.0))