Team System

def create_team_for_user(user_id: str, client=None):
    """
    Create a full AI Team for a specific user.
    
    Returns:
        tuple: (model, team)
    """
    model = create_model(user_id)
    
    # Create specialized sub-agents
    agents = [perplexity_agent, compound_agent, code_agent, context_qna_agent]
    
    # Create team leader
    team = Team(
        name="Hero Team",
        model=model,
        db=db,
        members=agents,
        tools=[BioTools(client=client), CalculatorTools()],
        instructions=get_prompt(),
        num_history_runs=AGENT_HISTORY_RUNS,
        enable_user_memories=True,
        memory_manager=memory_manager,
    )
    
    return model, team

team = Team(
    name="Hero Team",
    model=model,  # Main model (GPT-5, Claude, etc.)
    db=db,  # AsyncPostgresDb for session storage
    members=agents,  # List of specialized agents
    tools=[BioTools(client=client), CalculatorTools()],
    instructions=get_prompt(),  # System prompt from Phoenix
    num_history_runs=AGENT_HISTORY_RUNS,
    add_datetime_to_context=True,
    timezone_identifier="Asia/Kolkata",
    markdown=True,
    retries=AGENT_RETRIES,
    debug_mode=DEBUG_MODE,
    debug_level=DEBUG_LEVEL,
    enable_user_memories=True,
    memory_manager=memory_manager,  # Groq model for memory processing
)

def get_prompt() -> str:
    """Return system prompt content pulled from Phoenix or fallback."""
    prompt_name = "herocomp"
    
    try:
        fetched = client.prompts.get(prompt_identifier=prompt_name, tag="production")
        if hasattr(fetched, "format"):
            formatted = fetched.format()
        else:
            formatted = fetched
    except Exception as e:
        print("Phoenix prompt fetch error:", e)
        return get_system_prompt()  # Fallback
    
    messages = getattr(formatted, "messages", None)
    if not messages:
        return get_system_prompt()
    
    content = messages[0].get("content")
    return content or get_system_prompt()

# Groq for fast memory processing
memory_model = OpenAILike(
    id="openai/gpt-oss-120b",
    base_url="https://api.groq.com/openai/v1",
    api_key=GROQ_API_KEY,
)

memory_manager = MemoryManager(
    model=memory_model,
    db=db,  # Same AsyncPostgresDb
)

from collections import OrderedDict

_user_teams = OrderedDict()  # LRU cache
_user_team_locks: Dict[str, asyncio.Lock] = {}  # Per-user locks
_user_team_locks_guard = asyncio.Lock()  # Lock creation lock
_user_teams_lock = asyncio.Lock()  # Cache access lock

async def get_or_create_team(user_id: str, client=None):
    user_lock = await _get_user_team_lock(user_id)
    async with user_lock:
        async with _user_teams_lock:
            if user_id in _user_teams:
                # Move to end (mark as recently used)
                _user_teams.move_to_end(user_id)
                return _user_teams[user_id]
            
            # If cache full, evict oldest team
            if len(_user_teams) >= MAX_AGENTS:
                oldest_user, oldest_team = _user_teams.popitem(last=False)
        
        # Cleanup evicted team resources
        if oldest_team is not None:
            logger.info(f"[TeamCache] Evicting team for user {oldest_user}")
            await _remove_user_team_lock(oldest_user)
            # ... cleanup MCP connections ...
        
        # Create new team
        _, team = create_team_for_user(user_id, client=client)
        
        async with _user_teams_lock:
            _user_teams[user_id] = team
            cache_size = len(_user_teams)
        
        logger.info(f"[TeamCache] Created new team for user {user_id} (cache size: {cache_size}/{MAX_AGENTS})")
        return team

try:
    # Cleanup MCP connections if any
    if hasattr(oldest_team, 'members'):
        for member in oldest_team.members:
            if hasattr(member, 'tools'):
                for tool in member.tools:
                    if hasattr(tool, 'close'):
                        if hasattr(tool.close, '__await__'):
                            await tool.close()
                        else:
                            tool.close()
except Exception as e:
    logger.error(f"[TeamCache] Error during team cleanup: {e}", exc_info=True)

user_id = str(message.author.id)
session_id = str(message.channel.id)  # Channel = session

reply = await async_ask_junkie(
    prompt, user_id=user_id, session_id=session_id, images=images
)

if POSTGRES_URL:
    async_db_url = convert_to_async_url(POSTGRES_URL)
    db = AsyncPostgresDb(
        db_url=async_db_url,
        session_table="agent_sessions",  # Session/history storage
        memory_table="user_memories",    # User memory storage
    )
    logger.info("[DB] Using AsyncPostgresDb for session & memory storage")
else:
    db = None
    logger.warning("[DB] No POSTGRES_URL configured - sessions and memories will not persist!")

def convert_to_async_url(db_url: str) -> str:
    """Convert a postgresql:// URL to postgresql+asyncpg:// format."""
    if "+asyncpg" in db_url or "+psycopg_async" in db_url:
        return db_url
    
    try:
        parsed = make_url(db_url)
        async_url = f"postgresql+asyncpg://{parsed.username}:{parsed.password}@{parsed.host}"
        if parsed.port:
            async_url += f":{parsed.port}"
        async_url += f"/{parsed.database}"
        return async_url
    except Exception as e:
        logger.warning(f"Failed to convert URL: {e}")
        return db_url